diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json deleted file mode 100644 index 2a4e826..0000000 --- a/.devcontainer/devcontainer.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "name": "Python dev environment", - "image": "ghcr.io/opencyphal/toxic:tx22.4.2", - "workspaceFolder": "/workspace", - "workspaceMount": "source=${localWorkspaceFolder},target=/workspace,type=bind,consistency=delegated", - "mounts": [ - "source=root-vscode-server,target=/root/.vscode-server/extensions,type=volume", - "source=pydsdl-tox,target=/workspace/.nox,type=volume" - ], - "customizations": { - "vscode": { - "extensions": [ - "uavcan.dsdl", - "wholroyd.jinja", - "streetsidesoftware.code-spell-checker", - "ms-python.python", - "ms-python.mypy-type-checker", - "ms-python.black-formatter", - "ms-python.pylint" - ] - } - }, - "postCreateCommand": "git submodule update --init --recursive && tox -e local" -} diff --git a/.github/workflows/test-and-release.yml b/.github/workflows/test-and-release.yml index cc3a1cb..9efebdd 100644 --- a/.github/workflows/test-and-release.yml +++ b/.github/workflows/test-and-release.yml @@ -13,7 +13,7 @@ jobs: fail-fast: false matrix: os: [ ubuntu-latest, macos-latest ] - python: [ '3.8', '3.9', '3.10', '3.11', '3.12' ] + python: [ '3.7', '3.8', '3.9', '3.10', '3.11' ] include: - os: windows-2019 python: '3.10' @@ -60,6 +60,7 @@ jobs: echo "${{ runner.os }} not supported" exit 1 fi + python -c "import pydsdl; pydsdl.read_namespace('.dsdl-test/uavcan', [])" shell: bash pydsdl-release: diff --git a/.readthedocs.yml b/.readthedocs.yml index 3cebaef..97fb527 100644 --- a/.readthedocs.yml +++ b/.readthedocs.yml @@ -2,13 +2,6 @@ version: 2 -build: - os: ubuntu-22.04 - tools: - python: "3.12" - apt_packages: - - graphviz - sphinx: configuration: docs/conf.py fail_on_warning: true diff --git a/conftest.py b/conftest.py deleted file mode 100644 index d73f2d7..0000000 --- a/conftest.py +++ /dev/null @@ -1,71 +0,0 @@ -# -# Copyright (C) OpenCyphal Development Team -# Copyright Amazon.com Inc. or its affiliates. -# SPDX-License-Identifier: MIT -# -""" -Configuration for pytest tests including fixtures and hooks. -""" - -import tempfile -from pathlib import Path -from typing import Any, Optional - -import pytest - - -# +-------------------------------------------------------------------------------------------------------------------+ -# | TEST FIXTURES -# +-------------------------------------------------------------------------------------------------------------------+ -class TemporaryDsdlContext: - """ - Powers the temp_dsdl_factory test fixture. - """ - def __init__(self) -> None: - self._base_dir: Optional[Any] = None - - def new_file(self, file_path: Path, text: Optional[str] = None) -> Path: - if file_path.is_absolute(): - raise ValueError(f"{file_path} is an absolute path. The test fixture requires relative paths to work.") - file = self.base_dir / file_path - file.parent.mkdir(parents=True, exist_ok=True) - if text is not None: - file.write_text(text) - return file - - @property - def base_dir(self) -> Path: - if self._base_dir is None: - self._base_dir = tempfile.TemporaryDirectory() - return Path(self._base_dir.name).resolve() - - def _test_path_finalizer(self) -> None: - """ - Finalizer to clean up any temporary directories created during the test. - """ - if self._base_dir is not None: - self._base_dir.cleanup() - del self._base_dir - self._base_dir = None - -@pytest.fixture(scope="function") -def temp_dsdl_factory(request: pytest.FixtureRequest) -> Any: # pylint: disable=unused-argument - """ - Fixture for pydsdl tests that have to create files as part of the test. This object stays in-scope for a given - test method and does not requires a context manager in the test itself. - - Call `new_file(path)` to create a new file path in the fixture's temporary directory. This will create all - uncreated parent directories but will _not_ create the file unless text is provided: `new_file(path, "hello")` - """ - f = TemporaryDsdlContext() - request.addfinalizer(f._test_path_finalizer) # pylint: disable=protected-access - return f - - - -@pytest.fixture -def public_types() -> Path: - """ - Path to the public regulated data types directory used for tests. - """ - return Path(".dsdl-test") / "uavcan" diff --git a/docs/pages/pydsdl.rst b/docs/pages/pydsdl.rst index 9a9a7c4..4e73ac0 100644 --- a/docs/pages/pydsdl.rst +++ b/docs/pages/pydsdl.rst @@ -12,11 +12,10 @@ You can find a practical usage example in the Nunavut code generation library th :local: -The main functions -++++++++++++++++++ +The main function ++++++++++++++++++ .. autofunction:: pydsdl.read_namespace -.. autofunction:: pydsdl.read_files Type model diff --git a/docs/requirements.txt b/docs/requirements.txt index 3532611..9dc2a91 100644 --- a/docs/requirements.txt +++ b/docs/requirements.txt @@ -1,3 +1,3 @@ -sphinx >= 5.0 -sphinx_rtd_theme >= 1.0.0 +sphinx == 4.4.0 +sphinx_rtd_theme == 1.0.0 sphinx-computron >= 0.2, < 2.0 diff --git a/noxfile.py b/noxfile.py index b74b4c1..6939d8e 100644 --- a/noxfile.py +++ b/noxfile.py @@ -10,7 +10,7 @@ import nox -PYTHONS = ["3.8", "3.9", "3.10", "3.11", "3.12"] +PYTHONS = ["3.8", "3.9", "3.10", "3.11"] """The newest supported Python shall be listed LAST.""" nox.options.error_on_external_run = True @@ -33,6 +33,7 @@ def clean(session): "*.log", "*.tmp", ".nox", + ".dsdl-test", ] for w in wildcards: for f in Path.cwd().glob(w): @@ -48,9 +49,9 @@ def test(session): session.log("Using the newest supported Python: %s", is_latest_python(session)) session.install("-e", ".") session.install( - "pytest ~= 8.1", - "pytest-randomly ~= 3.15", - "coverage ~= 7.5", + "pytest ~= 7.3", + "pytest-randomly ~= 3.12", + "coverage ~= 7.2", ) session.run("coverage", "run", "-m", "pytest") session.run("coverage", "report", "--fail-under=95") @@ -60,7 +61,7 @@ def test(session): session.log(f"OPEN IN WEB BROWSER: file://{report_file}") -@nox.session(python=["3.8"]) +@nox.session(python=["3.7"]) def test_eol(session): """This is a minimal test session for those old Pythons that have EOLed.""" session.install("-e", ".") @@ -82,33 +83,30 @@ def pristine(session): @nox.session(python=PYTHONS, reuse_venv=True) def lint(session): - if is_oldest_python(session): - # we run mypy and pylint only on the oldest Python version to ensure maximum compatibility - session.install( - "mypy ~= 1.10", - "types-parsimonious", - "pylint ~= 3.2", - ) - session.run( - "mypy", - "--strict", - f"--config-file={ROOT_DIR / 'setup.cfg'}", - "pydsdl", - env={ - "MYPYPATH": str(THIRD_PARTY_DIR), - }, - ) - session.run( - "pylint", - str(ROOT_DIR / "pydsdl"), - env={ - "PYTHONPATH": str(THIRD_PARTY_DIR), - }, - ) + session.log("Using the newest supported Python: %s", is_latest_python(session)) + session.install( + "mypy ~= 1.2.0", + "pylint ~= 2.17.2", + ) + session.run( + "mypy", + "--strict", + f"--config-file={ROOT_DIR / 'setup.cfg'}", + "pydsdl", + env={ + "MYPYPATH": str(THIRD_PARTY_DIR), + }, + ) + session.run( + "pylint", + str(ROOT_DIR / "pydsdl"), + env={ + "PYTHONPATH": str(THIRD_PARTY_DIR), + }, + ) if is_latest_python(session): - # we run black only on the newest Python version to ensure that the code is formatted with the latest version - session.install("black ~= 24.4") - session.run("black", "--check", f"{ROOT_DIR / 'pydsdl'}") + session.install("black ~= 23.3") + session.run("black", "--check", ".") @nox.session(reuse_venv=True) @@ -123,6 +121,3 @@ def docs(session): def is_latest_python(session) -> bool: return PYTHONS[-1] in session.run("python", "-V", silent=True) - -def is_oldest_python(session) -> bool: - return PYTHONS[0] in session.run("python", "-V", silent=True) diff --git a/pydsdl/__init__.py b/pydsdl/__init__.py index aea7d00..d07f1f2 100644 --- a/pydsdl/__init__.py +++ b/pydsdl/__init__.py @@ -7,7 +7,7 @@ import sys as _sys from pathlib import Path as _Path -__version__ = "1.21.0" +__version__ = "1.20.1" __version_info__ = tuple(map(int, __version__.split(".")[:3])) __license__ = "MIT" __author__ = "OpenCyphal" @@ -25,9 +25,8 @@ _sys.path = [str(_Path(__file__).parent / "third_party")] + _sys.path # Never import anything that is not available here - API stability guarantees are only provided for the exposed items. -from ._dsdl import PrintOutputHandler as PrintOutputHandler from ._namespace import read_namespace as read_namespace -from ._namespace import read_files as read_files +from ._namespace import PrintOutputHandler as PrintOutputHandler # Error model. from ._error import FrontendError as FrontendError diff --git a/pydsdl/_bit_length_set/_symbolic_test.py b/pydsdl/_bit_length_set/_symbolic_test.py index 1c414d7..db262fa 100644 --- a/pydsdl/_bit_length_set/_symbolic_test.py +++ b/pydsdl/_bit_length_set/_symbolic_test.py @@ -2,6 +2,7 @@ # This software is distributed under the terms of the MIT License. # Author: Pavel Kirienko +import typing import random import itertools from ._symbolic import NullaryOperator, validate_numerically @@ -139,7 +140,7 @@ def _unittest_repetition() -> None: ) assert op.min == 7 * 3 assert op.max == 17 * 3 - assert set(op.expand()) == set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 3))) + assert set(op.expand()) == set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 3))) # type: ignore assert set(op.expand()) == {21, 25, 29, 31, 33, 35, 39, 41, 45, 51} assert set(op.modulo(7)) == {0, 1, 2, 3, 4, 5, 6} assert set(op.modulo(8)) == {1, 3, 5, 7} @@ -148,7 +149,7 @@ def _unittest_repetition() -> None: for _ in range(1): child = NullaryOperator(random.randint(0, 100) for _ in range(random.randint(1, 10))) k = random.randint(0, 10) - ref = set(map(sum, itertools.combinations_with_replacement(child.expand(), k))) + ref = set(map(sum, itertools.combinations_with_replacement(child.expand(), k))) # type: ignore op = RepetitionOperator(child, k) assert set(op.expand()) == ref @@ -156,7 +157,7 @@ def _unittest_repetition() -> None: assert op.max == max(child.expand()) * k div = random.randint(1, 64) - assert set(op.modulo(div)) == {x % div for x in ref} + assert set(op.modulo(div)) == {typing.cast(int, x) % div for x in ref} validate_numerically(op) @@ -172,9 +173,9 @@ def _unittest_range_repetition() -> None: assert op.max == 17 * 3 assert set(op.expand()) == ( {0} - | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 1))) - | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 2))) - | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 3))) + | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 1))) # type: ignore + | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 2))) # type: ignore + | set(map(sum, itertools.combinations_with_replacement([7, 11, 17], 3))) # type: ignore ) assert set(op.expand()) == {0, 7, 11, 14, 17, 18, 21, 22, 24, 25, 28, 29, 31, 33, 34, 35, 39, 41, 45, 51} assert set(op.modulo(7)) == {0, 1, 2, 3, 4, 5, 6} @@ -196,7 +197,10 @@ def _unittest_range_repetition() -> None: k_max = random.randint(0, 10) ref = set( itertools.chain( - *(map(sum, itertools.combinations_with_replacement(child.expand(), k)) for k in range(k_max + 1)) + *( + map(sum, itertools.combinations_with_replacement(child.expand(), k)) # type: ignore + for k in range(k_max + 1) + ) ) ) op = RangeRepetitionOperator(child, k_max) @@ -206,7 +210,7 @@ def _unittest_range_repetition() -> None: assert op.max == max(child.expand()) * k_max div = random.randint(1, 64) - assert set(op.modulo(div)) == {x % div for x in ref} + assert set(op.modulo(div)) == {typing.cast(int, x) % div for x in ref} validate_numerically(op) diff --git a/pydsdl/_data_type_builder.py b/pydsdl/_data_type_builder.py index 63b3ba4..4572da3 100644 --- a/pydsdl/_data_type_builder.py +++ b/pydsdl/_data_type_builder.py @@ -2,12 +2,16 @@ # This software is distributed under the terms of the MIT License. # Author: Pavel Kirienko +from typing import Optional, Callable, Iterable import logging from pathlib import Path -from typing import Callable, Iterable, Optional - -from . import _data_schema_builder, _error, _expression, _parser, _port_id_ranges, _serializable -from ._dsdl import DefinitionVisitor, DsdlFileBuildable +from . import _serializable +from . import _expression +from . import _error +from . import _dsdl_definition +from . import _parser +from . import _data_schema_builder +from . import _port_id_ranges class AssertionCheckFailureError(_error.InvalidDefinitionError): @@ -38,25 +42,21 @@ class MissingSerializationModeError(_error.InvalidDefinitionError): class DataTypeBuilder(_parser.StatementStreamProcessor): - - # pylint: disable=too-many-arguments def __init__( self, - definition: DsdlFileBuildable, - lookup_definitions: Iterable[DsdlFileBuildable], - definition_visitors: Iterable[DefinitionVisitor], + definition: _dsdl_definition.DSDLDefinition, + lookup_definitions: Iterable[_dsdl_definition.DSDLDefinition], print_output_handler: Callable[[int, str], None], allow_unregulated_fixed_port_id: bool, ): self._definition = definition self._lookup_definitions = list(lookup_definitions) - self._definition_visitors = definition_visitors self._print_output_handler = print_output_handler self._allow_unregulated_fixed_port_id = allow_unregulated_fixed_port_id self._element_callback = None # type: Optional[Callable[[str], None]] - assert isinstance(self._definition, DsdlFileBuildable) - assert all(map(lambda x: isinstance(x, DsdlFileBuildable), lookup_definitions)) + assert isinstance(self._definition, _dsdl_definition.DSDLDefinition) + assert all(map(lambda x: isinstance(x, _dsdl_definition.DSDLDefinition), lookup_definitions)) assert callable(self._print_output_handler) assert isinstance(self._allow_unregulated_fixed_port_id, bool) @@ -65,7 +65,7 @@ def __init__( def finalize(self) -> _serializable.CompositeType: if len(self._structs) == 1: # Structure type - (builder,) = self._structs + (builder,) = self._structs # type: _data_schema_builder.DataSchemaBuilder, out = self._make_composite( builder=builder, name=self._definition.full_name, @@ -198,7 +198,6 @@ def resolve_versioned_data_type(self, name: str, version: _serializable.Version) del name found = list(filter(lambda d: d.full_name == full_name and d.version == version, self._lookup_definitions)) if not found: - # Play Sherlock to help the user with mistakes like https://forum.opencyphal.org/t/904/2 requested_ns = full_name.split(_serializable.CompositeType.NAME_COMPONENT_SEPARATOR)[0] lookup_nss = set(x.root_namespace for x in self._lookup_definitions) @@ -222,20 +221,15 @@ def resolve_versioned_data_type(self, name: str, version: _serializable.Version) raise _error.InternalError("Conflicting definitions: %r" % found) target_definition = found[0] - for visitor in self._definition_visitors: - visitor(self._definition, target_definition) - - assert isinstance(target_definition, DsdlFileBuildable) + assert isinstance(target_definition, _dsdl_definition.DSDLDefinition) assert target_definition.full_name == full_name assert target_definition.version == version # Recursion is cool. - dt = target_definition.read( + return target_definition.read( lookup_definitions=self._lookup_definitions, - definition_visitors=self._definition_visitors, print_output_handler=self._print_output_handler, allow_unregulated_fixed_port_id=self._allow_unregulated_fixed_port_id, ) - return dt def _queue_attribute(self, element_callback: Callable[[str], None]) -> None: self._flush_attribute("") diff --git a/pydsdl/_dsdl.py b/pydsdl/_dsdl.py deleted file mode 100644 index 4e2f83b..0000000 --- a/pydsdl/_dsdl.py +++ /dev/null @@ -1,259 +0,0 @@ -# Copyright (C) OpenCyphal Development Team -# Copyright Amazon.com Inc. or its affiliates. -# SPDX-License-Identifier: MIT - -from abc import ABC, abstractmethod -from pathlib import Path -from typing import Any, Callable, Iterable, List, Optional, Set, Tuple, TypeVar, Union - -from ._serializable import CompositeType, Version - -PrintOutputHandler = Callable[[Path, int, str], None] -"""Invoked when the frontend encounters a print directive or needs to output a generic diagnostic.""" - - -class DsdlFile(ABC): - """ - Interface for DSDL files. This interface is used by the parser to abstract DSDL type details inferred from the - filesystem. Where properties are duplicated between the composite type and this file the composite type is to be - considered canonical. The properties directly on this class are inferred from the dsdl file path before the - composite type has been parsed. - """ - - @property - @abstractmethod - def composite_type(self) -> Optional[CompositeType]: - """The composite type that was read from the DSDL file or None if the type has not been parsed yet.""" - raise NotImplementedError() - - @property - @abstractmethod - def full_name(self) -> str: - """The full name, e.g., uavcan.node.Heartbeat""" - raise NotImplementedError() - - @property - def name_components(self) -> List[str]: - """Components of the full name as a list, e.g., ['uavcan', 'node', 'Heartbeat']""" - raise NotImplementedError() - - @property - @abstractmethod - def short_name(self) -> str: - """The last component of the full name, e.g., Heartbeat of uavcan.node.Heartbeat""" - raise NotImplementedError() - - @property - @abstractmethod - def full_namespace(self) -> str: - """The full name without the short name, e.g., uavcan.node for uavcan.node.Heartbeat""" - raise NotImplementedError() - - @property - @abstractmethod - def root_namespace(self) -> str: - """The first component of the full name, e.g., uavcan of uavcan.node.Heartbeat""" - raise NotImplementedError() - - @property - @abstractmethod - def text(self) -> str: - """The source text in its raw unprocessed form (with comments, formatting intact, and everything)""" - raise NotImplementedError() - - @property - @abstractmethod - def version(self) -> Version: - """ - The version of the DSDL definition. - """ - raise NotImplementedError() - - @property - @abstractmethod - def fixed_port_id(self) -> Optional[int]: - """Either the fixed port ID as integer, or None if not defined for this type.""" - raise NotImplementedError() - - @property - @abstractmethod - def has_fixed_port_id(self) -> bool: - """ - If the type has a fixed port ID defined, this method returns True. Equivalent to ``fixed_port_id is not None``. - """ - raise NotImplementedError() - - @property - @abstractmethod - def file_path(self) -> Path: - """The path to the DSDL file on the filesystem.""" - raise NotImplementedError() - - @property - @abstractmethod - def root_namespace_path(self) -> Path: - """ - The path to the root namespace directory on the filesystem. - """ - raise NotImplementedError() - - -class DsdlFileBuildable(DsdlFile): - """ - A DSDL file that can construct a composite type from its contents. - """ - - @abstractmethod - def read( - self, - lookup_definitions: Iterable["DsdlFileBuildable"], - definition_visitors: Iterable["DefinitionVisitor"], - print_output_handler: Callable[[int, str], None], - allow_unregulated_fixed_port_id: bool, - ) -> CompositeType: - """ - Reads the data type definition and returns its high-level data type representation. - The output should be cached; all following invocations should read from this cache. - Caching is very important, because it is expected that the same definition may be referred to multiple - times (e.g., for composition or when accessing external constants). Re-processing a definition every time - it is accessed would be a huge waste of time. - Note, however, that this may lead to unexpected complications if one is attempting to re-read a definition - with different inputs (e.g., different lookup paths) expecting to get a different result: caching would - get in the way. That issue is easy to avoid by creating a new instance of the object. - :param lookup_definitions: List of definitions available for referring to. - :param definition_visitors: Visitors to notify about discovered dependencies. - :param print_output_handler: Used for @print and for diagnostics: (line_number, text) -> None. - :param allow_unregulated_fixed_port_id: Do not complain about fixed unregulated port IDs. - :return: The data type representation. - """ - raise NotImplementedError() - - -DefinitionVisitor = Callable[[DsdlFile, DsdlFileBuildable], None] -""" -Called by the parser after if finds a dependent type but before it parses a file in a lookup namespace. -:param DsdlFile argument 0: The target DSDL file that has dependencies the parser is searching for. -:param DsdlFile argument 1: The dependency of target_dsdl_file file the parser is about to parse. -""" - -SortedFileT = TypeVar("SortedFileT", DsdlFile, DsdlFileBuildable, CompositeType) -SortedFileList = List[SortedFileT] -"""A list of DSDL files sorted by name, newest version first.""" - -FileSortKey: Callable[[SortedFileT], Tuple[str, int, int]] = lambda d: ( - d.full_name, - -d.version.major, - -d.version.minor, -) - - -def file_sort(file_list: Iterable[SortedFileT]) -> SortedFileList[SortedFileT]: - """ - Sorts a list of DSDL files lexicographically by name, newest version first. - """ - return list(sorted(file_list, key=FileSortKey)) - - -def normalize_paths_argument_to_list( - namespaces_or_namespace: Union[None, Path, str, Iterable[Union[Path, str]]], -) -> List[Path]: - """ - Normalizes the input argument to a list of paths. - """ - if namespaces_or_namespace is None: - return [] - if isinstance(namespaces_or_namespace, (Path, str)): - return [Path(namespaces_or_namespace)] - - def _convert(arg: Any) -> Path: - if not isinstance(arg, (str, Path)): - raise TypeError(f"Invalid type: {type(arg)}") - return Path(arg) if isinstance(arg, str) else arg - - return [_convert(arg) for arg in namespaces_or_namespace] - - -def normalize_paths_argument_to_set( - namespaces_or_namespace: Union[None, Path, str, Iterable[Union[Path, str]]], -) -> Set[Path]: - """ - Normalizes the input argument to a set of paths. - """ - return set(normalize_paths_argument_to_list(namespaces_or_namespace)) - - -# +-[UNIT TESTS]------------------------------------------------------------------------------------------------------+ - - -def _unittest_dsdl_normalize_paths_argument_to_list() -> None: - - from pytest import raises as assert_raises - - # Test with None argument - result = normalize_paths_argument_to_list(None) - assert result == [] - - # Test with single string argument - result = normalize_paths_argument_to_list("path/to/namespace") - assert result == [Path("path/to/namespace")] - - # Test with single Path argument - result = normalize_paths_argument_to_list(Path("path/to/namespace")) - assert result == [Path("path/to/namespace")] - - # Test with list of strings argument - result = normalize_paths_argument_to_list(["path/to/namespace1", "path/to/namespace2"]) - assert result == [Path("path/to/namespace1"), Path("path/to/namespace2")] - - # Test with list of Path arguments - result = normalize_paths_argument_to_list([Path("path/to/namespace1"), Path("path/to/namespace2")]) - assert result == [Path("path/to/namespace1"), Path("path/to/namespace2")] - - # Test with mixed list of strings and Path arguments - result = normalize_paths_argument_to_list(["path/to/namespace1", Path("path/to/namespace2")]) - assert result == [Path("path/to/namespace1"), Path("path/to/namespace2")] - - # Test with invalid argument type - with assert_raises(TypeError): - normalize_paths_argument_to_list(42) # type: ignore - - # Test with invalid argument type - with assert_raises(TypeError): - normalize_paths_argument_to_list([42]) # type: ignore - - -def _unittest_dsdl_normalize_paths_argument_to_set() -> None: - - from pytest import raises as assert_raises - - # Test with None argument - result = normalize_paths_argument_to_set(None) - assert result == set() - - # Test with single string argument - result = normalize_paths_argument_to_set("path/to/namespace") - assert result == {Path("path/to/namespace")} - - # Test with single Path argument - result = normalize_paths_argument_to_set(Path("path/to/namespace")) - assert result == {Path("path/to/namespace")} - - # Test with list of strings argument - result = normalize_paths_argument_to_set(["path/to/namespace1", "path/to/namespace2"]) - assert result == {Path("path/to/namespace1"), Path("path/to/namespace2")} - - # Test with list of Path arguments - result = normalize_paths_argument_to_set([Path("path/to/namespace1"), Path("path/to/namespace2")]) - assert result == {Path("path/to/namespace1"), Path("path/to/namespace2")} - - # Test with mixed list of strings and Path arguments - result = normalize_paths_argument_to_set(["path/to/namespace1", Path("path/to/namespace2")]) - assert result == {Path("path/to/namespace1"), Path("path/to/namespace2")} - - # Test with invalid argument type - with assert_raises(TypeError): - normalize_paths_argument_to_set(42) # type: ignore - - # Test with invalid argument type - with assert_raises(TypeError): - normalize_paths_argument_to_set([42]) # type: ignore diff --git a/pydsdl/_dsdl_definition.py b/pydsdl/_dsdl_definition.py index f8f7dad..a8114da 100644 --- a/pydsdl/_dsdl_definition.py +++ b/pydsdl/_dsdl_definition.py @@ -2,16 +2,14 @@ # This software is distributed under the terms of the MIT License. # Author: Pavel Kirienko -import logging import time +from typing import Iterable, Callable, Optional, List +import logging from pathlib import Path -from typing import Callable, Iterable, List, Optional - -from . import _parser -from ._data_type_builder import DataTypeBuilder -from ._dsdl import DefinitionVisitor, DsdlFileBuildable -from ._error import FrontendError, InternalError, InvalidDefinitionError +from ._error import FrontendError, InvalidDefinitionError, InternalError from ._serializable import CompositeType, Version +from . import _parser + _logger = logging.getLogger(__name__) @@ -25,7 +23,7 @@ def __init__(self, text: str, path: Path): super().__init__(text=text, path=Path(path)) -class DSDLDefinition(DsdlFileBuildable): +class DSDLDefinition: """ A DSDL type definition source abstracts the filesystem level details away, presenting a higher-level interface that operates solely on the level of type names, namespaces, fixed identifiers, and so on. @@ -38,18 +36,15 @@ def __init__(self, file_path: Path, root_namespace_path: Path): del file_path self._root_namespace_path = Path(root_namespace_path) del root_namespace_path - self._text: Optional[str] = None + with open(self._file_path) as f: + self._text = str(f.read()) # Checking the sanity of the root directory path - can't contain separators if CompositeType.NAME_COMPONENT_SEPARATOR in self._root_namespace_path.name: raise FileNameFormatError("Invalid namespace name", path=self._root_namespace_path) # Determining the relative path within the root namespace directory - try: - relative_path = self._root_namespace_path.name / self._file_path.relative_to(self._root_namespace_path) - except ValueError: - # the file is not under the same root path so we'll have to make an assumption that the - relative_path = Path(self._root_namespace_path.name) / self._file_path.name + relative_path = self._root_namespace_path.name / self._file_path.relative_to(self._root_namespace_path) # Parsing the basename, e.g., 434.GetTransportStatistics.0.1.dsdl basename_components = relative_path.name.split(".")[:-1] @@ -91,24 +86,31 @@ def __init__(self, file_path: Path, root_namespace_path: Path): self._cached_type: Optional[CompositeType] = None - # +-----------------------------------------------------------------------+ - # | DsdlFileBuildable :: INTERFACE | - # +-----------------------------------------------------------------------+ def read( self, - lookup_definitions: Iterable[DsdlFileBuildable], - definition_visitors: Iterable[DefinitionVisitor], + lookup_definitions: Iterable["DSDLDefinition"], print_output_handler: Callable[[int, str], None], allow_unregulated_fixed_port_id: bool, ) -> CompositeType: + """ + Reads the data type definition and returns its high-level data type representation. + The output is cached; all following invocations will read from the cache. + Caching is very important, because it is expected that the same definition may be referred to multiple + times (e.g., for composition or when accessing external constants). Re-processing a definition every time + it is accessed would be a huge waste of time. + Note, however, that this may lead to unexpected complications if one is attempting to re-read a definition + with different inputs (e.g., different lookup paths) expecting to get a different result: caching would + get in the way. That issue is easy to avoid by creating a new instance of the object. + :param lookup_definitions: List of definitions available for referring to. + :param print_output_handler: Used for @print and for diagnostics: (line_number, text) -> None. + :param allow_unregulated_fixed_port_id: Do not complain about fixed unregulated port IDs. + :return: The data type representation. + """ log_prefix = "%s.%d.%d" % (self.full_name, self.version.major, self.version.minor) if self._cached_type is not None: _logger.debug("%s: Cache hit", log_prefix) return self._cached_type - if not self._file_path.exists(): - raise InvalidDefinitionError("Attempt to read DSDL file that doesn't exist.", self._file_path) - started_at = time.monotonic() # Remove the target definition from the lookup list in order to prevent @@ -122,17 +124,17 @@ def read( ", ".join(set(sorted(map(lambda x: x.root_namespace, lookup_definitions)))), ) try: - builder = DataTypeBuilder( + builder = _data_type_builder.DataTypeBuilder( definition=self, lookup_definitions=lookup_definitions, - definition_visitors=definition_visitors, print_output_handler=print_output_handler, allow_unregulated_fixed_port_id=allow_unregulated_fixed_port_id, ) - - _parser.parse(self.text, builder) + with open(self.file_path) as f: + _parser.parse(f.read(), builder) self._cached_type = builder.finalize() + _logger.info( "%s: Processed in %.0f ms; category: %s, fixed port ID: %s", log_prefix, @@ -149,38 +151,34 @@ def read( except Exception as ex: # pragma: no cover raise InternalError(culprit=ex, path=self.file_path) from ex - # +-----------------------------------------------------------------------+ - # | DsdlFile :: INTERFACE | - # +-----------------------------------------------------------------------+ - @property - def composite_type(self) -> Optional[CompositeType]: - return self._cached_type - @property def full_name(self) -> str: + """The full name, e.g., uavcan.node.Heartbeat""" return self._name @property def name_components(self) -> List[str]: + """Components of the full name as a list, e.g., ['uavcan', 'node', 'Heartbeat']""" return self._name.split(CompositeType.NAME_COMPONENT_SEPARATOR) @property def short_name(self) -> str: + """The last component of the full name, e.g., Heartbeat of uavcan.node.Heartbeat""" return self.name_components[-1] @property def full_namespace(self) -> str: + """The full name without the short name, e.g., uavcan.node for uavcan.node.Heartbeat""" return str(CompositeType.NAME_COMPONENT_SEPARATOR.join(self.name_components[:-1])) @property def root_namespace(self) -> str: + """The first component of the full name, e.g., uavcan of uavcan.node.Heartbeat""" return self.name_components[0] @property def text(self) -> str: - if self._text is None: - with open(self._file_path) as f: - self._text = str(f.read()) + """The source text in its raw unprocessed form (with comments, formatting intact, and everything)""" return self._text @property @@ -189,6 +187,7 @@ def version(self) -> Version: @property def fixed_port_id(self) -> Optional[int]: + """Either the fixed port ID as integer, or None if not defined for this type.""" return self._fixed_port_id @property @@ -203,12 +202,6 @@ def file_path(self) -> Path: def root_namespace_path(self) -> Path: return self._root_namespace_path - # +-----------------------------------------------------------------------+ - # | Python :: SPECIAL FUNCTIONS | - # +-----------------------------------------------------------------------+ - def __hash__(self) -> int: - return hash((self.full_name, self.version)) - def __eq__(self, other: object) -> bool: """ Two definitions will compare equal if they share the same name AND version number. @@ -229,25 +222,6 @@ def __str__(self) -> str: __repr__ = __str__ -# +-[UNIT TESTS]------------------------------------------------------------------------------------------------------+ - - -def _unittest_dsdl_definition_read_non_existant() -> None: - from pytest import raises as expect_raises - - target = Path("root", "ns", "Target.1.1.dsdl") - target_definition = DSDLDefinition(target, target.parent) - - def print_output(line_number: int, text: str) -> None: # pragma: no cover - pass - - with expect_raises(InvalidDefinitionError): - target_definition.read([], [], print_output, True) - - -def _unittest_dsdl_definition_read_text(temp_dsdl_factory) -> None: # type: ignore - target_root = Path("root", "ns") - target_file_path = Path(target_root / "Target.1.1.dsdl") - dsdl_file = temp_dsdl_factory.new_file(target_root / target_file_path, "@sealed") - target_definition = DSDLDefinition(dsdl_file, target_root) - assert "@sealed" == target_definition.text +# Moved this import here to break recursive dependency. +# Maybe I have messed up the architecture? Should think about it later. +from . import _data_type_builder # pylint: disable=wrong-import-position diff --git a/pydsdl/_error.py b/pydsdl/_error.py index 283486f..d301765 100644 --- a/pydsdl/_error.py +++ b/pydsdl/_error.py @@ -108,9 +108,6 @@ class InvalidDefinitionError(FrontendError): """ -# +-[UNIT TESTS]------------------------------------------------------------------------------------------------------+ - - def _unittest_error() -> None: try: raise FrontendError("Hello world!") @@ -127,8 +124,8 @@ def _unittest_error() -> None: try: raise FrontendError("Hello world!", path=Path("path/to/file.dsdl")) except Exception as ex: - assert repr(ex) == "FrontendError: 'path/to/file.dsdl: Hello world!'" assert str(ex) == "path/to/file.dsdl: Hello world!" + assert repr(ex) == "FrontendError: 'path/to/file.dsdl: Hello world!'" def _unittest_internal_error_github_reporting() -> None: @@ -153,7 +150,7 @@ def _unittest_internal_error_github_reporting() -> None: print(ex) assert ex.path == Path("FILE_PATH") assert ex.line == 42 - # We have to ignore the last couple of characters because Python before 3.7 repr's Exceptions like this: + # We have to ignore the last couple of characters because Python before 3.7 reprs Exceptions like this: # Exception('ERROR TEXT',) # But newer Pythons do it like this: # Exception('ERROR TEXT') diff --git a/pydsdl/_namespace.py b/pydsdl/_namespace.py index dd77c43..8e9501e 100644 --- a/pydsdl/_namespace.py +++ b/pydsdl/_namespace.py @@ -4,19 +4,13 @@ # pylint: disable=logging-not-lazy -import collections +from typing import Iterable, Callable, DefaultDict, List, Optional, Union, Set, Dict import logging -from itertools import product, repeat +import collections from pathlib import Path -from typing import Callable, DefaultDict, Dict, Iterable, List, Optional, Set, Tuple, Union - -from . import _dsdl_definition, _error, _serializable -from ._dsdl import DsdlFileBuildable, PrintOutputHandler, SortedFileList -from ._dsdl import file_sort as dsdl_file_sort -from ._dsdl import normalize_paths_argument_to_list, normalize_paths_argument_to_set -from ._namespace_reader import DsdlDefinitions, read_definitions - -_logger = logging.getLogger(__name__) +from . import _serializable +from . import _dsdl_definition +from . import _error class RootNamespaceNameCollisionError(_error.InvalidDefinitionError): @@ -75,13 +69,8 @@ class SealingConsistencyError(_error.InvalidDefinitionError): """ -class DsdlPathInferenceError(_error.InvalidDefinitionError): - """ - Raised when the namespace, type, fixed port ID, or version cannot be inferred from a file path. - """ - - -# +--[PUBLIC API]-----------------------------------------------------------------------------------------------------+ +PrintOutputHandler = Callable[[Path, int, str], None] +"""Invoked when the frontend encounters a print directive or needs to output a generic diagnostic.""" def read_namespace( @@ -92,7 +81,7 @@ def read_namespace( allow_root_namespace_name_collision: bool = True, ) -> List[_serializable.CompositeType]: """ - This function is a main entry point for the library. + This function is the main entry point of the library. It reads all DSDL definitions from the specified root namespace directory and produces the annotated AST. :param root_namespace_directory: The path of the root namespace directory that will be read. @@ -119,26 +108,48 @@ def read_namespace( the same root namespace name multiple times in the lookup dirs. This will enable defining a namespace partially and let other entities define new messages or new sub-namespaces in the same root namespace. - :return: A list of :class:`pydsdl.CompositeType` found under the `root_namespace_directory` and sorted - lexicographically by full data type name, then by major version (newest version first), then by minor - version (newest version first). The ordering guarantee allows the caller to always find the newest version - simply by picking the first matching occurrence. + :return: A list of :class:`pydsdl.CompositeType` sorted lexicographically by full data type name, + then by major version (newest version first), then by minor version (newest version first). + The ordering guarantee allows the caller to always find the newest version simply by picking + the first matching occurrence. :raises: :class:`pydsdl.FrontendError`, :class:`MemoryError`, :class:`SystemError`, :class:`OSError` if directories do not exist or inaccessible, :class:`ValueError`/:class:`TypeError` if the arguments are invalid. """ + # Add the own root namespace to the set of lookup directories, sort lexicographically, remove duplicates. + # We'd like this to be an iterable list of strings but we handle the common practice of passing in a single path. + if lookup_directories is None: + lookup_directories_path_list: List[Path] = [] + elif isinstance(lookup_directories, (str, bytes, Path)): + lookup_directories_path_list = [Path(lookup_directories)] + else: + lookup_directories_path_list = list(map(Path, lookup_directories)) + + for a in lookup_directories_path_list: + if not isinstance(a, (str, Path)): + raise TypeError("Lookup directories shall be an iterable of paths. Found in list: " + type(a).__name__) + _logger.debug(_LOG_LIST_ITEM_PREFIX + str(a)) + # Normalize paths and remove duplicates. Resolve symlinks to avoid ambiguities. root_namespace_directory = Path(root_namespace_directory).resolve() + lookup_directories_path_list.append(root_namespace_directory) + lookup_directories_path_list = list(sorted({x.resolve() for x in lookup_directories_path_list})) + _logger.debug("Lookup directories are listed below:") + for a in lookup_directories_path_list: + _logger.debug(_LOG_LIST_ITEM_PREFIX + str(a)) - lookup_directories_path_list = _construct_lookup_directories_path_list( - [root_namespace_directory], - normalize_paths_argument_to_list(lookup_directories), - allow_root_namespace_name_collision, - ) + # Check for common usage errors and warn the user if anything looks suspicious. + _ensure_no_common_usage_errors(root_namespace_directory, lookup_directories_path_list, _logger.warning) + + # Check the namespaces. + _ensure_no_nested_root_namespaces(lookup_directories_path_list) + + if not allow_root_namespace_name_collision: + _ensure_no_namespace_name_collisions(lookup_directories_path_list) # Construct DSDL definitions from the target and the lookup dirs. - target_dsdl_definitions = _construct_dsdl_definitions_from_namespaces([root_namespace_directory]) + target_dsdl_definitions = _construct_dsdl_definitions_from_namespace(root_namespace_directory) if not target_dsdl_definitions: _logger.info("The namespace at %s is empty", root_namespace_directory) return [] @@ -146,124 +157,9 @@ def read_namespace( for x in target_dsdl_definitions: _logger.debug(_LOG_LIST_ITEM_PREFIX + str(x)) - return _complete_read_function( - target_dsdl_definitions, lookup_directories_path_list, print_output_handler, allow_unregulated_fixed_port_id - ).direct - - -# pylint: disable=too-many-arguments -def read_files( - dsdl_files: Union[None, Path, str, Iterable[Union[Path, str]]], - root_namespace_directories_or_names: Union[None, Path, str, Iterable[Union[Path, str]]], - lookup_directories: Union[None, Path, str, Iterable[Union[Path, str]]] = None, - print_output_handler: Optional[PrintOutputHandler] = None, - allow_unregulated_fixed_port_id: bool = False, -) -> Tuple[List[_serializable.CompositeType], List[_serializable.CompositeType]]: - """ - This function is a main entry point for the library. - It reads all DSDL definitions from the specified ``dsdl_files`` and produces the annotated AST for these types and - the transitive closure of the types they depend on. - - :param dsdl_files: A list of paths to dsdl files to parse. - - :param root_namespace_directories_or_names: This can be a set of names of root namespaces or relative paths to - root namespaces. All ``dsdl_files`` provided must be under one of these roots. For example, given: - - .. code-block:: python - - dsdl_files = [ - Path("workspace/project/types/animals/felines/Tabby.1.0.dsdl"), - Path("workspace/project/types/animals/canines/Boxer.1.0.dsdl"), - Path("workspace/project/types/plants/trees/DouglasFir.1.0.dsdl") - ] - - - then this argument must be one of: - - .. code-block:: python - - root_namespace_directories_or_names = ["animals", "plants"] - - root_namespace_directories_or_names = [ - Path("workspace/project/types/animals"), - Path("workspace/project/types/plants") - ] - - - :param lookup_directories: List of other namespace directories containing data type definitions that are - referred to from the target dsdl files. For example, if you are reading vendor-specific types, - the list of lookup directories should always include a path to the standard root namespace ``uavcan``, - otherwise the types defined in the vendor-specific namespace won't be able to use data types from the - standard namespace. - - :param print_output_handler: If provided, this callable will be invoked when a ``@print`` directive - is encountered or when the frontend needs to emit a diagnostic; - the arguments are: path, line number (1-based), text. - If not provided, no output will be produced except for the standard Python logging subsystem - (but ``@print`` expressions will be evaluated anyway, and a failed evaluation will be a fatal error). - - :param allow_unregulated_fixed_port_id: Do not reject unregulated fixed port identifiers. - As demanded by the specification, the frontend rejects unregulated fixed port ID by default. - This is a dangerous feature that must not be used unless you understand the risks. - Please read https://opencyphal.org/guide. - - :return: A Tuple of lists of :class:`pydsdl.CompositeType`. The first index in the Tuple are the types parsed from - the ``dsdl_files`` argument. The second index are types that the target ``dsdl_files`` utilizes. - A note for using these values to describe build dependencies: each :class:`pydsdl.CompositeType` has two - fields that provide links back to the filesystem where the dsdl files were located when parsing the type; - ``source_file_path`` and ``source_file_path_to_root``. - - :raises: :class:`pydsdl.FrontendError`, :class:`MemoryError`, :class:`SystemError`, - :class:`OSError` if directories do not exist or inaccessible, - :class:`ValueError`/:class:`TypeError` if the arguments are invalid. - """ - # Normalize paths and remove duplicates. Resolve symlinks to avoid ambiguities. - target_dsdl_definitions = _construct_dsdl_definitions_from_files( - normalize_paths_argument_to_list(dsdl_files), - normalize_paths_argument_to_set(root_namespace_directories_or_names), - ) - if len(target_dsdl_definitions) == 0: - _logger.info("No DSDL files found in the specified directories") - return ([], []) - - if _logger.isEnabledFor(logging.DEBUG): # pragma: no cover - _logger.debug("Target DSDL definitions are listed below:") - - for x in target_dsdl_definitions: - _logger.debug(_LOG_LIST_ITEM_PREFIX + str(x.file_path)) - - root_namespaces = {f.root_namespace_path.resolve() for f in target_dsdl_definitions} - lookup_directories_path_list = _construct_lookup_directories_path_list( - root_namespaces, - normalize_paths_argument_to_list(lookup_directories), - True, - ) - - definitions = _complete_read_function( - target_dsdl_definitions, lookup_directories_path_list, print_output_handler, allow_unregulated_fixed_port_id - ) - - return (definitions.direct, definitions.transitive) - - -# +--[INTERNAL API::PUBLIC API HELPERS]-------------------------------------------------------------------------------+ -# These are functions called by the public API before the actual processing begins. - -DSDL_FILE_SUFFIX = ".dsdl" -DSDL_FILE_GLOB = f"*{DSDL_FILE_SUFFIX}" -DSDL_FILE_SUFFIX_LEGACY = ".uavcan" -DSDL_FILE_GLOB_LEGACY = f"*{DSDL_FILE_SUFFIX_LEGACY}" -_LOG_LIST_ITEM_PREFIX = " " * 4 - - -def _complete_read_function( - target_dsdl_definitions: SortedFileList[DsdlFileBuildable], - lookup_directories_path_list: List[Path], - print_output_handler: Optional[PrintOutputHandler], - allow_unregulated_fixed_port_id: bool, -) -> DsdlDefinitions: - - lookup_dsdl_definitions = _construct_dsdl_definitions_from_namespaces(lookup_directories_path_list) + lookup_dsdl_definitions = [] # type: List[_dsdl_definition.DSDLDefinition] + for ld in lookup_directories_path_list: + lookup_dsdl_definitions += _construct_dsdl_definitions_from_namespace(ld) # Check for collisions against the lookup definitions also. _ensure_no_collisions(target_dsdl_definitions, lookup_dsdl_definitions) @@ -281,9 +177,8 @@ def _complete_read_function( ", ".join(set(sorted(map(lambda t: t.root_namespace, lookup_dsdl_definitions)))), ) - # This is the biggie. All the rest of the wranging is just to get to this point. This will take the - # most time and memory. - definitions = read_definitions( + # Read the constructed definitions. + types = _read_namespace_definitions( target_dsdl_definitions, lookup_dsdl_definitions, print_output_handler, allow_unregulated_fixed_port_id ) @@ -293,107 +188,62 @@ def _complete_read_function( # directories may contain issues and mistakes that are outside of the control of the user (e.g., # they could be managed by a third party) -- the user shouldn't be affected by mistakes committed # by the third party. - _ensure_no_fixed_port_id_collisions(definitions.direct) - _ensure_minor_version_compatibility(definitions.transitive + definitions.direct) + _ensure_no_fixed_port_id_collisions(types) + _ensure_minor_version_compatibility(types) - return definitions - - -def _construct_lookup_directories_path_list( - root_namespace_directories: Iterable[Path], - lookup_directories_path_list: List[Path], - allow_root_namespace_name_collision: bool, -) -> List[Path]: - """ - Intermediate transformation and validation of inputs into a list of lookup directories as paths. + return types - :param root_namespace_directory: The path of the root namespace directory that will be read. - For example, ``dsdl/uavcan`` to read the ``uavcan`` namespace. - :param lookup_directories: List of other namespace directories containing data type definitions that are - referred to from the target root namespace. For example, if you are reading a vendor-specific namespace, - the list of lookup directories should always include a path to the standard root namespace ``uavcan``, - otherwise the types defined in the vendor-specific namespace won't be able to use data types from the - standard namespace. +DSDL_FILE_GLOB = "*.dsdl" +DSDL_FILE_GLOB_LEGACY = "*.uavcan" +_LOG_LIST_ITEM_PREFIX = " " * 4 - :param allow_root_namespace_name_collision: Allow using the source root namespace name in the look up dirs or - the same root namespace name multiple times in the lookup dirs. This will enable defining a namespace - partially and let other entities define new messages or new sub-namespaces in the same root namespace. +_logger = logging.getLogger(__name__) - :return: A list of lookup directories as paths. - :raises: :class:`pydsdl.FrontendError`, :class:`MemoryError`, :class:`SystemError`, - :class:`OSError` if directories do not exist or inaccessible, - :class:`ValueError`/:class:`TypeError` if the arguments are invalid. +def _read_namespace_definitions( + target_definitions: List[_dsdl_definition.DSDLDefinition], + lookup_definitions: List[_dsdl_definition.DSDLDefinition], + print_output_handler: Optional[PrintOutputHandler] = None, + allow_unregulated_fixed_port_id: bool = False, +) -> List[_serializable.CompositeType]: + """ + Construct type descriptors from the specified target definitions. + Allow the target definitions to use the lookup definitions within themselves. + :param target_definitions: Which definitions to read. + :param lookup_definitions: Which definitions can be used by the processed definitions. + :return: A list of types. """ - # Add the own root namespace to the set of lookup directories, sort lexicographically, remove duplicates. - # We'd like this to be an iterable list of strings but we handle the common practice of passing in a single path. - - # Normalize paths and remove duplicates. Resolve symlinks to avoid ambiguities. - lookup_directories_path_list.extend(root_namespace_directories) - lookup_directories_path_list = list(sorted({x.resolve() for x in lookup_directories_path_list})) - _logger.debug("Lookup directories are listed below:") - for a in lookup_directories_path_list: - _logger.debug(_LOG_LIST_ITEM_PREFIX + str(a)) - - # Check for common usage errors and warn the user if anything looks suspicious. - _ensure_no_common_usage_errors(root_namespace_directories, lookup_directories_path_list, _logger.warning) - - # Check the namespaces and ensure that there are no name collisions. - _ensure_no_namespace_name_collisions_or_nested_root_namespaces( - lookup_directories_path_list, allow_root_namespace_name_collision - ) - - return lookup_directories_path_list - - -def _construct_dsdl_definitions_from_files( - dsdl_files: List[Path], - valid_roots: Set[Path], -) -> SortedFileList[DsdlFileBuildable]: - """ """ - output = set() # type: Set[DsdlFileBuildable] - for fp in dsdl_files: - root_namespace_path = _infer_path_to_root(fp, valid_roots) - if fp.suffix == DSDL_FILE_SUFFIX_LEGACY: - _logger.warning( - "File uses deprecated extension %r, please rename to use %r: %s", - DSDL_FILE_SUFFIX_LEGACY, - DSDL_FILE_SUFFIX, - fp, - ) - output.add(_dsdl_definition.DSDLDefinition(fp, root_namespace_path)) - return dsdl_file_sort(output) + def make_print_handler(definition: _dsdl_definition.DSDLDefinition) -> Callable[[int, str], None]: + def handler(line_number: int, text: str) -> None: + if print_output_handler: # pragma: no branch + assert isinstance(line_number, int) and isinstance(text, str) + assert line_number > 0, "Line numbers must be one-based" + print_output_handler(definition.file_path, line_number, text) + return handler -def _construct_dsdl_definitions_from_namespaces( - root_namespace_paths: List[Path], -) -> SortedFileList[DsdlFileBuildable]: - """ - Accepts a directory path, returns a sorted list of abstract DSDL file representations. Those can be read later. - The definitions are sorted by name lexicographically, then by major version (greatest version first), - then by minor version (same ordering as the major version). - """ - source_file_paths: Set[Tuple[Path, Path]] = set() # index of all file paths already found - for root_namespace_path in root_namespace_paths: - for p in root_namespace_path.rglob(DSDL_FILE_GLOB): - source_file_paths.add((p, root_namespace_path)) - for p in root_namespace_path.rglob(DSDL_FILE_GLOB_LEGACY): - source_file_paths.add((p, root_namespace_path)) - _logger.warning( - "File uses deprecated extension %r, please rename to use %r: %s", - DSDL_FILE_GLOB_LEGACY, - DSDL_FILE_GLOB, - p, - ) + types = [] # type: List[_serializable.CompositeType] + for tdd in target_definitions: + try: + dt = tdd.read(lookup_definitions, make_print_handler(tdd), allow_unregulated_fixed_port_id) + except _error.FrontendError as ex: # pragma: no cover + ex.set_error_location_if_unknown(path=tdd.file_path) + raise ex + except (MemoryError, SystemError): # pragma: no cover + raise + except Exception as ex: # pragma: no cover + raise _error.InternalError(culprit=ex, path=tdd.file_path) from ex + else: + types.append(dt) - return dsdl_file_sort([_dsdl_definition.DSDLDefinition(*p) for p in source_file_paths]) + return types def _ensure_no_collisions( - target_definitions: List[DsdlFileBuildable], - lookup_definitions: List[DsdlFileBuildable], + target_definitions: List[_dsdl_definition.DSDLDefinition], + lookup_definitions: List[_dsdl_definition.DSDLDefinition], ) -> None: for tg in target_definitions: tg_full_namespace_period = tg.full_namespace.lower() + "." @@ -525,7 +375,7 @@ def _ensure_minor_version_compatibility_pairwise( def _ensure_no_common_usage_errors( - root_namespace_directories: Iterable[Path], lookup_directories: Iterable[Path], reporter: Callable[[str], None] + root_namespace_directory: Path, lookup_directories: Iterable[Path], reporter: Callable[[str], None] ) -> None: suspicious_base_names = [ "public_regulated_data_types", @@ -541,7 +391,7 @@ def is_valid_name(s: str) -> bool: return True # resolve() will also normalize the case in case-insensitive filesystems. - all_paths = {y.resolve() for y in root_namespace_directories} | {x.resolve() for x in lookup_directories} + all_paths = {root_namespace_directory.resolve()} | {x.resolve() for x in lookup_directories} for p in all_paths: try: candidates = [x for x in p.iterdir() if x.is_dir() and is_valid_name(x.name)] @@ -558,91 +408,59 @@ def is_valid_name(s: str) -> bool: reporter(report) -def _ensure_no_namespace_name_collisions_or_nested_root_namespaces( - directories: Iterable[Path], allow_name_collisions: bool -) -> None: - directories = {x.resolve() for x in directories} # normalize the case in case-insensitive filesystems - - def check_each(path_tuple_with_result: Tuple[Tuple[Path, Path], List[int]]) -> bool: - path_tuple = path_tuple_with_result[0] - if not path_tuple[0].samefile(path_tuple[1]): - if not allow_name_collisions and path_tuple[0].name.lower() == path_tuple[1].name.lower(): - return True +def _ensure_no_nested_root_namespaces(directories: Iterable[Path]) -> None: + dirs = {x.resolve() for x in directories} # normalize the case in case-insensitive filesystems + for a in dirs: + for b in dirs: + if a.samefile(b): + continue try: - path_tuple[0].relative_to(path_tuple[1]) + a.relative_to(b) except ValueError: pass else: - path_tuple_with_result[1][0] = 1 - return True - return False - - # zip a list[1] of int 0 so we can assign a failure type. 0 is name collision and 1 is nested root namespace - # further cartesian checks can be added here using this pattern - - # next/filter returns the first failure or None if no failures - check_result = next(filter(check_each, zip(product(directories, directories), repeat([0]))), None) - - if check_result: - path_tuple = check_result[0] - failure_type = check_result[1][0] - if failure_type == 0: - raise RootNamespaceNameCollisionError( - "The following namespaces have the same name: %s" % path_tuple[0], path=path_tuple[1] - ) - else: - raise NestedRootNamespaceError( - "The following namespace is nested inside this one, which is not permitted: %s" % path_tuple[0], - path=path_tuple[1], - ) + raise NestedRootNamespaceError( + "The following namespace is nested inside this one, which is not permitted: %s" % a, path=b + ) -def _infer_path_to_root(dsdl_path: Path, valid_dsdl_roots_or_path_to_root: Set[Path]) -> Path: +def _ensure_no_namespace_name_collisions(directories: Iterable[Path]) -> None: + directories = {x.resolve() for x in directories} # normalize the case in case-insensitive filesystems + for a in directories: + for b in directories: + if a.samefile(b): + continue + if a.name.lower() == b.name.lower(): + _logger.info("Collision: %r [%r] == %r [%r]", a, a.name, b, b.name) + raise RootNamespaceNameCollisionError("The name of this namespace conflicts with %s" % b, path=a) + + +def _construct_dsdl_definitions_from_namespace(root_namespace_path: Path) -> List[_dsdl_definition.DSDLDefinition]: """ - Infer the path to the namespace root of a DSDL file path. - :param dsdl_path: The path to the alleged DSDL file. - :param valid_dsdl_roots_or_path_to_root: The set of valid root names or paths under which the type must reside. - :return The path to the root namespace directory. - :raises DsdlPathInferenceError: If the namespace root cannot be inferred from the provided information. + Accepts a directory path, returns a sorted list of abstract DSDL file representations. Those can be read later. + The definitions are sorted by name lexicographically, then by major version (greatest version first), + then by minor version (same ordering as the major version). """ - if valid_dsdl_roots_or_path_to_root is None: - raise _error.InternalError("valid_dsdl_roots_or_path_to_root was None") - - if dsdl_path.is_absolute() and len(valid_dsdl_roots_or_path_to_root) == 0: - raise DsdlPathInferenceError( - f"dsdl_path ({dsdl_path}) is absolute and the provided valid root names are empty. The DSDL root of " - "an absolute path cannot be inferred without this information.", + source_file_paths: Set[Path] = set() + for p in root_namespace_path.rglob(DSDL_FILE_GLOB): + source_file_paths.add(p) + for p in root_namespace_path.rglob(DSDL_FILE_GLOB_LEGACY): + source_file_paths.add(p) + _logger.warning( + "File uses deprecated extension %r, please rename to use %r: %s", DSDL_FILE_GLOB_LEGACY, DSDL_FILE_GLOB, p ) - if len(valid_dsdl_roots_or_path_to_root) == 0: - # if we have no valid roots we can only infer the root of the path. We require the path to be relative - # to avoid accidental inferences given that dsdl file trees starting from a filesystem root are rare. - return Path(dsdl_path.parts[0]) - - # The strongest inference is when the path is relative to a known root. - for path_to_root in valid_dsdl_roots_or_path_to_root: - try: - _ = dsdl_path.relative_to(path_to_root) - except ValueError: - continue - return path_to_root - - # A weaker, but valid inference is when the path is a child of a known root folder name. - root_parts = {x.parts[-1] for x in valid_dsdl_roots_or_path_to_root if len(x.parts) == 1} - parts = list(dsdl_path.parent.parts) - for i, part in list(enumerate(parts)): - if part in root_parts: - return Path().joinpath(*parts[: i + 1]) - # +1 to include the root folder - raise DsdlPathInferenceError(f"No valid root found in path {str(dsdl_path)}") + output = [] # type: List[_dsdl_definition.DSDLDefinition] + for fp in sorted(source_file_paths): + dsdl_def = _dsdl_definition.DSDLDefinition(fp, root_namespace_path) + output.append(dsdl_def) - -# +--[ UNIT TESTS ]---------------------------------------------------------------------------------------------------+ + # Lexicographically by name, newest version first. + return list(sorted(output, key=lambda d: (d.full_name, -d.version.major, -d.version.minor))) def _unittest_dsdl_definition_constructor() -> None: import tempfile - from ._dsdl_definition import FileNameFormatError with tempfile.TemporaryDirectory() as directory: @@ -654,9 +472,9 @@ def _unittest_dsdl_definition_constructor() -> None: (root / "nested/2.Asd.21.32.dsdl").write_text("# TEST B") (root / "nested/Foo.32.43.dsdl").write_text("# TEST C") - dsdl_defs = _construct_dsdl_definitions_from_namespaces([root]) + dsdl_defs = _construct_dsdl_definitions_from_namespace(root) print(dsdl_defs) - lut = {x.full_name: x for x in dsdl_defs} # type: Dict[str, DsdlFileBuildable] + lut = {x.full_name: x for x in dsdl_defs} # type: Dict[str, _dsdl_definition.DSDLDefinition] assert len(lut) == 3 assert str(lut["foo.Qwerty"]) == repr(lut["foo.Qwerty"]) @@ -710,7 +528,7 @@ def _unittest_dsdl_definition_constructor() -> None: (root / "nested/Malformed.MAJOR.MINOR.dsdl").touch() try: - _construct_dsdl_definitions_from_namespaces([root]) + _construct_dsdl_definitions_from_namespace(root) except FileNameFormatError as ex: print(ex) (root / "nested/Malformed.MAJOR.MINOR.dsdl").unlink() @@ -719,7 +537,7 @@ def _unittest_dsdl_definition_constructor() -> None: (root / "nested/NOT_A_NUMBER.Malformed.1.0.dsdl").touch() try: - _construct_dsdl_definitions_from_namespaces([root]) + _construct_dsdl_definitions_from_namespace(root) except FileNameFormatError as ex: print(ex) (root / "nested/NOT_A_NUMBER.Malformed.1.0.dsdl").unlink() @@ -728,26 +546,26 @@ def _unittest_dsdl_definition_constructor() -> None: (root / "nested/Malformed.dsdl").touch() try: - _construct_dsdl_definitions_from_namespaces([root]) + _construct_dsdl_definitions_from_namespace(root) except FileNameFormatError as ex: print(ex) (root / "nested/Malformed.dsdl").unlink() else: # pragma: no cover assert False - _construct_dsdl_definitions_from_namespaces([root]) # making sure all errors are cleared + _construct_dsdl_definitions_from_namespace(root) # making sure all errors are cleared (root / "nested/super.bad").mkdir() (root / "nested/super.bad/Unreachable.1.0.dsdl").touch() try: - _construct_dsdl_definitions_from_namespaces([root]) + _construct_dsdl_definitions_from_namespace(root) except FileNameFormatError as ex: print(ex) else: # pragma: no cover assert False try: - _construct_dsdl_definitions_from_namespaces([root / "nested/super.bad"]) + _construct_dsdl_definitions_from_namespace(root / "nested/super.bad") except FileNameFormatError as ex: print(ex) else: # pragma: no cover @@ -764,9 +582,9 @@ def _unittest_dsdl_definition_constructor_legacy() -> None: root = di / "foo" root.mkdir() (root / "123.Qwerty.123.234.uavcan").write_text("# TEST A") - dsdl_defs = _construct_dsdl_definitions_from_namespaces([root]) + dsdl_defs = _construct_dsdl_definitions_from_namespace(root) print(dsdl_defs) - lut = {x.full_name: x for x in dsdl_defs} # type: Dict[str, DsdlFileBuildable] + lut = {x.full_name: x for x in dsdl_defs} # type: Dict[str, _dsdl_definition.DSDLDefinition] assert len(lut) == 1 t = lut["foo.Qwerty"] assert t.file_path == root / "123.Qwerty.123.234.uavcan" @@ -791,33 +609,32 @@ def _unittest_common_usage_errors() -> None: reports = [] # type: List[str] - _ensure_no_common_usage_errors([root_ns_dir], [], reports.append) + _ensure_no_common_usage_errors(root_ns_dir, [], reports.append) assert not reports - _ensure_no_common_usage_errors([root_ns_dir], [di / "baz"], reports.append) + _ensure_no_common_usage_errors(root_ns_dir, [di / "baz"], reports.append) assert not reports dir_dsdl = root_ns_dir / "dsdl" dir_dsdl.mkdir() - _ensure_no_common_usage_errors([dir_dsdl], [di / "baz"], reports.append) + _ensure_no_common_usage_errors(dir_dsdl, [di / "baz"], reports.append) assert not reports # Because empty. dir_dsdl_vscode = dir_dsdl / ".vscode" dir_dsdl_vscode.mkdir() - _ensure_no_common_usage_errors([dir_dsdl], [di / "baz"], reports.append) + _ensure_no_common_usage_errors(dir_dsdl, [di / "baz"], reports.append) assert not reports # Because the name is not valid. dir_dsdl_uavcan = dir_dsdl / "uavcan" dir_dsdl_uavcan.mkdir() - _ensure_no_common_usage_errors([dir_dsdl], [di / "baz"], reports.append) + _ensure_no_common_usage_errors(dir_dsdl, [di / "baz"], reports.append) (rep,) = reports reports.clear() assert str(dir_dsdl_uavcan.resolve()).lower() in rep.lower() def _unittest_nested_roots() -> None: - import tempfile - from pytest import raises + import tempfile with tempfile.TemporaryDirectory() as directory: di = Path(directory) @@ -826,13 +643,13 @@ def _unittest_nested_roots() -> None: (di / "a/b").mkdir() (di / "a/c").mkdir() (di / "aa/b").mkdir() - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([], True) - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([di / "a"], True) - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([di / "a/b", di / "a/c"], True) + _ensure_no_nested_root_namespaces([]) + _ensure_no_nested_root_namespaces([di / "a"]) + _ensure_no_nested_root_namespaces([di / "a/b", di / "a/c"]) with raises(NestedRootNamespaceError): - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([di / "a/b", di / "a"], True) - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([di / "aa/b", di / "a"], True) - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([di / "a/b", di / "aa"], True) + _ensure_no_nested_root_namespaces([di / "a/b", di / "a"]) + _ensure_no_nested_root_namespaces([di / "aa/b", di / "a"]) + _ensure_no_nested_root_namespaces([di / "a/b", di / "aa"]) def _unittest_issue_71() -> None: # https://github.com/OpenCyphal/pydsdl/issues/71 @@ -846,165 +663,3 @@ def _unittest_issue_71() -> None: # https://github.com/OpenCyphal/pydsdl/issues (real / "Msg.0.1.dsdl").write_text("@sealed") assert len(read_namespace(real, [real, link])) == 1 assert len(read_namespace(link, [real, link])) == 1 - - -def _unittest_type_from_path_inference() -> None: - from pytest import raises as expect_raises - - # To determine the namespace do - - dsdl_file = Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl") - path_to_root = _infer_path_to_root(dsdl_file, {Path("/repo/uavcan")}) - namespace_parts = dsdl_file.parent.relative_to(path_to_root.parent).parts - - assert path_to_root == Path("/repo/uavcan") - assert namespace_parts == ("uavcan", "foo", "bar") - - # The simplest inference made is when relative dsdl paths are provided with no additional information. In this - # case the method assumes that the relative path is the correct and complete namespace of the type: - - # relative path - root = _infer_path_to_root(Path("uavcan/foo/bar/435.baz.1.0.dsdl"), set()) - assert root == Path("uavcan") - - # The root namespace is not inferred in an absolute path without additional data: - - with expect_raises(DsdlPathInferenceError): - _ = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), set()) - - with expect_raises(_error.InternalError): - _ = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), None) # type: ignore - - # If an absolute path is provided along with a path-to-root "hint" then the former must be relative to the - # latter: - - # dsdl file path is not contained within the root path - with expect_raises(DsdlPathInferenceError): - _ = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("/not-a-repo")}) - - root = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("/repo/uavcan")}) - assert root == Path("/repo/uavcan") - - # The priority is given to paths that are relative to the root when both simple root names and paths are provided: - root = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("foo"), Path("/repo/uavcan")}) - assert root == Path("/repo/uavcan") - - root = _infer_path_to_root(Path("repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("foo"), Path("repo/uavcan")}) - assert root == Path("repo/uavcan") - - # Finally, the method will infer the root namespace from simple folder names if no additional information is - # provided: - - valid_roots = {Path("uavcan"), Path("cyphal")} - - # absolute dsdl path using valid roots - root = _infer_path_to_root(Path("/repo/uavcan/foo/bar/435.baz.1.0.dsdl"), valid_roots) - assert root == Path("/repo/uavcan") - - # relative dsdl path using valid roots - root = _infer_path_to_root(Path("repo/uavcan/foo/bar/435.baz.1.0.dsdl"), valid_roots) - assert root == Path("repo/uavcan") - - # absolute dsdl path using valid roots but an invalid file path - with expect_raises(DsdlPathInferenceError): - _ = _infer_path_to_root(Path("/repo/crap/foo/bar/435.baz.1.0.dsdl"), valid_roots) - - # relative dsdl path using valid roots but an invalid file path - with expect_raises(DsdlPathInferenceError): - _ = _infer_path_to_root(Path("repo/crap/foo/bar/435.baz.1.0.dsdl"), valid_roots) - - # relative dsdl path with invalid root fragments - invalid_root_fragments = {Path("cyphal", "acme")} - with expect_raises(DsdlPathInferenceError): - _ = _infer_path_to_root(Path("repo/crap/foo/bar/435.baz.1.0.dsdl"), invalid_root_fragments) - - # In this example, foo/bar might look like a valid root path but it is not relative to repo/uavcan/foo/bar and is - # not considered after relative path inference has failed because it is not a simple root name. - root = _infer_path_to_root(Path("repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("foo/bar"), Path("foo")}) - assert root == Path("repo/uavcan/foo") - - # when foo/bar is placed within the proper, relative path it is considered as a valid root and is preferred over - # the simple root name "foo": - root = _infer_path_to_root(Path("repo/uavcan/foo/bar/435.baz.1.0.dsdl"), {Path("repo/uavcan/foo/bar"), Path("foo")}) - assert root == Path("repo/uavcan/foo/bar") - - -def _unittest_type_read_files_example(temp_dsdl_factory) -> None: # type: ignore - - # let's test the comments for the read function - dsdl_files = [ - Path("workspace/project/types/animals/felines/Tabby.1.0.uavcan"), # keep .uavcan to cover the warning - Path("workspace/project/types/animals/canines/Boxer.1.0.dsdl"), - Path("workspace/project/types/plants/trees/DouglasFir.1.0.dsdl"), - ] - - dsdl_files_abs = [] - root_namespace_paths = set() - for dsdl_file in dsdl_files: - dsdl_files_abs.append(temp_dsdl_factory.new_file(dsdl_file, "@sealed")) - root_namespace_paths.add(temp_dsdl_factory.base_dir / dsdl_file.parent.parent) - root_namespace_directories_or_names_simple = ["animals", "plants"] - - direct, transitive = read_files(dsdl_files_abs, root_namespace_directories_or_names_simple) - - assert len(direct) == len(dsdl_files) - assert len(transitive) == 0 - - for direct_type in direct: - assert direct_type.root_namespace in root_namespace_directories_or_names_simple - assert direct_type.source_file_path_to_root in root_namespace_paths - - direct, _ = read_files(dsdl_files_abs, root_namespace_paths) - - assert len(direct) == len(dsdl_files) - - for direct_type in direct: - assert direct_type.root_namespace in root_namespace_directories_or_names_simple - assert direct_type.source_file_path_to_root in root_namespace_paths - - -def _unittest_targets_found_in_lookup_namespaces(temp_dsdl_factory) -> None: # type: ignore - - # call read_files with a list of dsdl files which are also located in the provided lookup namespaces - - plant_1_0 = Path("types/plants/Plant.1.0.dsdl") - tree_1_0 = Path("types/plants/trees/Tree.1.0.dsdl") - douglas_fir_1_0 = Path("types/plants/trees/DouglasFir.1.0.dsdl") - - plant_file = temp_dsdl_factory.new_file(plant_1_0, "@sealed\n") - test_files = [ - temp_dsdl_factory.new_file(tree_1_0, "@sealed\nplants.Plant.1.0 plt\n"), - temp_dsdl_factory.new_file(douglas_fir_1_0, "@sealed\nplants.trees.Tree.1.0 tree\n"), - ] - lookup_dirs = [plant_file.parent] - - direct, transitive = read_files(test_files, lookup_dirs) - - assert len(direct) == len(test_files) - assert len(transitive) == 1 - - -def _unittest_read_files_empty_args() -> None: - direct, transitive = read_files([], []) - - assert len(direct) == 0 - assert len(transitive) == 0 - - -def _unittest_ensure_no_collisions(temp_dsdl_factory) -> None: # type: ignore - from pytest import raises as expect_raises - - # gratuitous coverage of the collision check where other tests don't cover some edge cases - _ensure_no_namespace_name_collisions_or_nested_root_namespaces([], False) - - with expect_raises(DataTypeNameCollisionError): - _ensure_no_collisions( - [_dsdl_definition.DSDLDefinition(Path("a/b.1.0.dsdl"), Path("a"))], - [_dsdl_definition.DSDLDefinition(Path("a/B.1.0.dsdl"), Path("a"))], - ) - - with expect_raises(DataTypeNameCollisionError): - _ensure_no_collisions( - [_dsdl_definition.DSDLDefinition(Path("a/b/c.1.0.dsdl"), Path("a"))], - [_dsdl_definition.DSDLDefinition(Path("a/b.1.0.dsdl"), Path("a"))], - ) diff --git a/pydsdl/_namespace_reader.py b/pydsdl/_namespace_reader.py deleted file mode 100644 index ee271ca..0000000 --- a/pydsdl/_namespace_reader.py +++ /dev/null @@ -1,387 +0,0 @@ -# Copyright (C) OpenCyphal Development Team -# Copyright Amazon.com Inc. or its affiliates. -# SPDX-License-Identifier: MIT - - -import functools -import logging -from pathlib import Path -from typing import Dict, List, NamedTuple, Optional, Set, cast - -from ._dsdl import DsdlFile, DsdlFileBuildable, PrintOutputHandler, SortedFileList -from ._dsdl import file_sort as dsdl_file_sort -from ._error import FrontendError, InternalError -from ._serializable._composite import CompositeType - - -# pylint: disable=too-many-arguments -def _read_definitions( - target_definitions: SortedFileList[DsdlFileBuildable], - lookup_definitions: SortedFileList[DsdlFileBuildable], - print_output_handler: Optional[PrintOutputHandler], - allow_unregulated_fixed_port_id: bool, - direct: Set[CompositeType], - transitive: Set[CompositeType], - file_pool: Dict[Path, DsdlFileBuildable], - level: int, -) -> None: - """ - Don't look at me! I'm hideous! - (recursive method with a lot of arguments. See read_definitions for documentation) - """ - - _pending_definitions: Set[DsdlFileBuildable] = set() - - def callback(_: DsdlFile, dependent_type: DsdlFileBuildable) -> None: - if dependent_type.file_path not in file_pool: - _pending_definitions.add(dependent_type) - - def print_handler(file: Path, line: int, message: str) -> None: - if print_output_handler is not None: - print_output_handler(file, line, message) - - for target_definition in target_definitions: - - if not isinstance(target_definition, DsdlFileBuildable): - raise TypeError("Expected DsdlFileBuildable, got: " + type(target_definition).__name__) - - target_definition = file_pool.setdefault(target_definition.file_path, target_definition) - # make sure we are working with the same object for a given file path - - if target_definition.composite_type is not None and ( - target_definition.composite_type in direct or target_definition.composite_type in transitive - ): - logging.debug("Skipping target file %s because it has already been processed", target_definition.file_path) - if level == 0 and target_definition.composite_type in transitive: - # promote to direct - transitive.remove(target_definition.composite_type) - direct.add(target_definition.composite_type) - continue - - try: - new_composite_type = target_definition.read( - lookup_definitions, - [callback], - functools.partial(print_handler, target_definition.file_path), - allow_unregulated_fixed_port_id, - ) - except FrontendError as ex: # pragma: no cover - ex.set_error_location_if_unknown(path=target_definition.file_path) - raise ex - except Exception as ex: # pragma: no cover - raise InternalError(culprit=ex, path=target_definition.file_path) from ex - - if level == 0: - - direct.add(new_composite_type) - try: - transitive.remove(new_composite_type) - except KeyError: - pass - else: - transitive.add(new_composite_type) - - if len(_pending_definitions) > 0: - _read_definitions( - dsdl_file_sort(_pending_definitions), - lookup_definitions, - print_output_handler, - allow_unregulated_fixed_port_id, - direct, - transitive, - file_pool, - level + 1, - ) - _pending_definitions.clear() - - -# +---[FILE: PUBLIC]--------------------------------------------------------------------------------------------------+ - -DsdlDefinitions = NamedTuple( - "DsdlDefinitions", [("direct", SortedFileList[CompositeType]), ("transitive", SortedFileList[CompositeType])] -) -""" -Common DSDL definition set including the direct dependencies requested and the transitive dependencies found. The former -and latter sets will be disjoint. -""" - - -def read_definitions( - target_definitions: SortedFileList[DsdlFileBuildable], - lookup_definitions: SortedFileList[DsdlFileBuildable], - print_output_handler: Optional[PrintOutputHandler], - allow_unregulated_fixed_port_id: bool, -) -> DsdlDefinitions: - """ - Given a set of DSDL files, this method reads the text and invokes the parser for each and for any files found in the - lookup set where these are used by the target set. - - :param target_definitions: List of definitions to read. - :param lookup_definitions: List of definitions available for referring to. - :param print_output_handler: Used for @print and for diagnostics: (line_number, text) -> None. - :param allow_unregulated_fixed_port_id: Do not complain about fixed unregulated port IDs. - :return: The data type representation. - :raises InvalidDefinitionError: If a dependency is missing. - :raises InternalError: If an unexpected error occurs. - """ - _direct: Set[CompositeType] = set() - _transitive: Set[CompositeType] = set() - _file_pool: Dict[Path, DsdlFileBuildable] = {} - _read_definitions( - target_definitions, - lookup_definitions, - print_output_handler, - allow_unregulated_fixed_port_id, - _direct, - _transitive, - _file_pool, - 0, - ) - return DsdlDefinitions( - dsdl_file_sort(_direct), - dsdl_file_sort(_transitive), - ) - - -# +-[UNIT TESTS]------------------------------------------------------------------------------------------------------+ - - -def _unittest_namespace_reader_read_definitions(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - target = temp_dsdl_factory.new_file(Path("root", "ns", "Target.1.1.dsdl"), "@sealed") - target_definitions = [cast(DsdlFileBuildable, _dsdl_definition.DSDLDefinition(target, target.parent))] - lookup_definitions: List[DsdlFileBuildable] = [] - - read_definitions(target_definitions, lookup_definitions, None, True) - - -def _unittest_namespace_reader_read_definitions_multiple(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "Target.1.1.dsdl"), "@sealed\nns.Aisle.1.0 paper_goods\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "Target.2.0.dsdl"), "@sealed\nns.Aisle.2.0 paper_goods\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "Walmart.2.4.dsdl"), "@sealed\nns.Aisle.1.0 paper_goods\n"), - ] - aisles = [ - temp_dsdl_factory.new_file(Path("root", "ns", "Aisle.1.0.dsdl"), "@sealed"), - temp_dsdl_factory.new_file(Path("root", "ns", "Aisle.2.0.dsdl"), "@sealed"), - temp_dsdl_factory.new_file(Path("root", "ns", "Aisle.3.0.dsdl"), "@sealed"), - ] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(a, a.parent) for a in aisles], - None, - True, - ) - - assert len(definitions.direct) == 3 - assert len(definitions.transitive) == 2 - - -def _unittest_namespace_reader_read_definitions_multiple_no_load(temp_dsdl_factory) -> None: # type: ignore - """ - Ensure that the loader does not load files that are not in the transitive closure of the target files. - """ - from . import _dsdl_definition - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "Adams.1.0.dsdl"), "@sealed\nns.Tacoma.1.0 volcano\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "Hood.1.0.dsdl"), "@sealed\nns.Rainer.1.0 volcano\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "StHelens.2.1.dsdl"), "@sealed\nns.Baker.1.0 volcano\n"), - ] - dependencies = [ - temp_dsdl_factory.new_file(Path("root", "ns", "Tacoma.1.0.dsdl"), "@sealed"), - temp_dsdl_factory.new_file(Path("root", "ns", "Rainer.1.0.dsdl"), "@sealed"), - temp_dsdl_factory.new_file(Path("root", "ns", "Baker.1.0.dsdl"), "@sealed"), - Path( - "root", "ns", "Shasta.1.0.dsdl" - ), # since this isn't in the transitive closure of target dependencies it will - # never be read thus it will not be an error that it does not exist. - ] - - target_definitions = [cast(DsdlFileBuildable, _dsdl_definition.DSDLDefinition(t, t.parent)) for t in targets] - lookup_definitions = [cast(DsdlFileBuildable, _dsdl_definition.DSDLDefinition(a, a.parent)) for a in dependencies] - _ = read_definitions( - target_definitions, - lookup_definitions, - None, - True, - ) - - # make sure Shasta.1.0 was never accessed but Tacoma 1.0 was - last_item = lookup_definitions[-1] - assert isinstance(last_item, _dsdl_definition.DSDLDefinition) - assert last_item._text is None # pylint: disable=protected-access - assert lookup_definitions[0].composite_type is not None - - # Make sure text is cached. - assert lookup_definitions[0].text == lookup_definitions[0].text - - -def _unittest_namespace_reader_read_definitions_promotion(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - user_1_0 = temp_dsdl_factory.new_file(Path("root", "ns", "User.1.0.dsdl"), "@sealed\n") - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "User.2.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - user_1_0, - ] - lookups = [user_1_0] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(l, l.parent) for l in lookups], - None, - True, - ) - - assert len(definitions.direct) == 2 - assert len(definitions.transitive) == 0 - - -def _unittest_namespace_reader_read_definitions_no_demote(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - user_1_0 = temp_dsdl_factory.new_file(Path("root", "ns", "User.1.0.dsdl"), "@sealed\n") - targets = [ - user_1_0, - temp_dsdl_factory.new_file(Path("root", "ns", "User.2.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - ] - lookups = [user_1_0] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(l, l.parent) for l in lookups], - None, - True, - ) - - assert len(definitions.direct) == 2 - assert len(definitions.transitive) == 0 - -def _unittest_namespace_reader_read_definitions_no_promote(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "User.2.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "User.3.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - ] - lookups = [temp_dsdl_factory.new_file(Path("root", "ns", "User.1.0.dsdl"), "@sealed\n")] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(l, l.parent) for l in lookups], - None, - True, - ) - - assert len(definitions.direct) == 2 - assert len(definitions.transitive) == 1 - -def _unittest_namespace_reader_read_definitions_twice(temp_dsdl_factory) -> None: # type: ignore - from . import _dsdl_definition - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "User.2.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "User.2.0.dsdl"), "@sealed\nns.User.1.0 old_guy\n"), - ] - lookups = [temp_dsdl_factory.new_file(Path("root", "ns", "User.1.0.dsdl"), "@sealed\n")] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(l, l.parent) for l in lookups], - None, - True, - ) - - assert len(definitions.direct) == 1 - assert len(definitions.transitive) == 1 - -def _unittest_namespace_reader_read_definitions_missing_dependency(temp_dsdl_factory) -> None: # type: ignore - """ - Verify that an error is raised when a dependency is missing. - """ - from pytest import raises as assert_raises - - from . import _dsdl_definition - from ._data_type_builder import UndefinedDataTypeError - - with assert_raises(UndefinedDataTypeError): - read_definitions( - [ - _dsdl_definition.DSDLDefinition( - f := temp_dsdl_factory.new_file( - Path("root", "ns", "Cat.1.0.dsdl"), "@sealed\nns.Birman.1.0 fluffy\n" - ), - f.parent, - ) - ], - [], - None, - True, - ) - - -def _unittest_namespace_reader_read_definitions_target_in_lookup(temp_dsdl_factory) -> None: # type: ignore - """ - Ensure the direct and transitive sets are disjoint. - """ - from . import _dsdl_definition - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "Ontario.1.0.dsdl"), "@sealed\nns.NewBrunswick.1.0 place\n"), - temp_dsdl_factory.new_file(Path("root", "ns", "NewBrunswick.1.0.dsdl"), "@sealed"), - ] - lookup = [ - temp_dsdl_factory.new_file(Path("root", "ns", "NewBrunswick.1.0.dsdl"), "@sealed"), - ] - - definitions = read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [_dsdl_definition.DSDLDefinition(l, l.parent) for l in lookup], - None, - True, - ) - - assert len(definitions.direct) == 2 - assert len(definitions.transitive) == 0 - - -def _unittest_namespace_reader_read_defs_target_dont_allow_unregulated(temp_dsdl_factory) -> None: # type: ignore - """ - Ensure that an error is raised when an invalid, fixed port ID is used without an override. - """ - from pytest import raises as assert_raises - - from . import _dsdl_definition - from ._data_type_builder import UnregulatedFixedPortIDError - - targets = [ - temp_dsdl_factory.new_file(Path("root", "ns", "845.Lice.1.0.dsdl"), "@sealed\n"), - ] - - with assert_raises(UnregulatedFixedPortIDError): - read_definitions( - [_dsdl_definition.DSDLDefinition(t, t.parent) for t in targets], - [], - None, - False, - ) - - -def _unittest_namespace_reader_type_error() -> None: - - from pytest import raises as assert_raises - - from . import _dsdl_definition - - with assert_raises(TypeError): - read_definitions( - [""], # type: ignore - [], - None, - True, - ) diff --git a/pydsdl/_parser.py b/pydsdl/_parser.py index 4eda81a..b73a533 100644 --- a/pydsdl/_parser.py +++ b/pydsdl/_parser.py @@ -8,7 +8,7 @@ import functools import fractions from pathlib import Path -from typing import cast, List, Tuple +from typing import List, Tuple import parsimonious from parsimonious.nodes import Node as _Node from . import _error @@ -263,11 +263,11 @@ def visit_type_version_specifier(self, _n: _Node, children: _Children) -> _seria return _serializable.Version(major=major.as_native_integer(), minor=minor.as_native_integer()) def visit_type_primitive_truncated(self, _n: _Node, children: _Children) -> _serializable.PrimitiveType: - _kw, _sp, cons = cast(Tuple[_Node, _Node, _PrimitiveTypeConstructor], children) + _kw, _sp, cons = children # type: _Node, _Node, _PrimitiveTypeConstructor return cons(_serializable.PrimitiveType.CastMode.TRUNCATED) def visit_type_primitive_saturated(self, _n: _Node, children: _Children) -> _serializable.PrimitiveType: - _, cons = cast(Tuple[_Node, _PrimitiveTypeConstructor], children) + _, cons = children # type: _Node, _PrimitiveTypeConstructor return cons(_serializable.PrimitiveType.CastMode.SATURATED) def visit_type_primitive_name_boolean(self, _n: _Node, _c: _Children) -> _PrimitiveTypeConstructor: diff --git a/pydsdl/_serializable/_composite.py b/pydsdl/_serializable/_composite.py index 1153cc7..6ed71da 100644 --- a/pydsdl/_serializable/_composite.py +++ b/pydsdl/_serializable/_composite.py @@ -5,15 +5,17 @@ import abc import math import typing +import itertools from pathlib import Path - -from .. import _expression, _port_id_ranges +from .. import _expression +from .. import _port_id_ranges from .._bit_length_set import BitLengthSet -from ._attribute import Attribute, Constant, Field, PaddingField -from ._name import InvalidNameError, check_name -from ._primitive import PrimitiveType, UnsignedIntegerType from ._serializable import SerializableType, TypeParameterError +from ._attribute import Attribute, Field, PaddingField, Constant +from ._name import check_name, InvalidNameError from ._void import VoidType +from ._primitive import PrimitiveType, UnsignedIntegerType + Version = typing.NamedTuple("Version", [("major", int), ("minor", int)]) @@ -53,7 +55,7 @@ class CompositeType(SerializableType): MAX_VERSION_NUMBER = 255 NAME_COMPONENT_SEPARATOR = "." - def __init__( # pylint: disable=too-many-arguments, too-many-locals + def __init__( # pylint: disable=too-many-arguments self, name: str, version: Version, @@ -95,26 +97,9 @@ def __init__( # pylint: disable=too-many-arguments, too-many-locals "Name is too long: %r is longer than %d characters" % (self._name, self.MAX_NAME_LENGTH) ) - self._name_components = self._name.split(self.NAME_COMPONENT_SEPARATOR) - for component in self._name_components: + for component in self._name.split(self.NAME_COMPONENT_SEPARATOR): check_name(component) - def search_up_for_root(path: Path, namespace_components: typing.List[str]) -> Path: - if namespace_components[-1] != path.stem: - raise InvalidNameError( - f"{path.stem} != {namespace_components[-1]}. Source file directory structure " - f"is not consistent with the type's namespace ({self._name_components}, " - f"{self._source_file_path})" - ) - if len(namespace_components) == 1: - return path - return search_up_for_root(path.parent, namespace_components[:-1]) - - self._path_to_root_namespace = search_up_for_root( - self._source_file_path.parent, - (self.namespace_components if not self._has_parent_service else self.namespace_components[:-1]), - ) - # Version check version_valid = ( (0 <= self._version.major <= self.MAX_VERSION_NUMBER) @@ -163,12 +148,7 @@ def full_name(self) -> str: @property def name_components(self) -> typing.List[str]: """Components of the full name as a list, e.g., ``['uavcan', 'node', 'Heartbeat']``.""" - return self._name_components - - @property - def namespace_components(self) -> typing.List[str]: - """Components of the namspace as a list, e.g., ``['uavcan', 'node']``.""" - return self._name_components[:-1] + return self._name.split(CompositeType.NAME_COMPONENT_SEPARATOR) @property def short_name(self) -> str: @@ -183,7 +163,7 @@ def doc(self) -> str: @property def full_namespace(self) -> str: """The full name without the short name, e.g., ``uavcan.node`` for ``uavcan.node.Heartbeat``.""" - return str(CompositeType.NAME_COMPONENT_SEPARATOR.join(self.namespace_components)) + return str(CompositeType.NAME_COMPONENT_SEPARATOR.join(self.name_components[:-1])) @property def root_namespace(self) -> str: @@ -259,30 +239,10 @@ def has_fixed_port_id(self) -> bool: @property def source_file_path(self) -> Path: """ - The path to the dsdl file from which this type was read. - For synthesized types such as service request/response sections, this property is the path to the service type - since request and response types are defined within the service type's dsdl file. + For synthesized types such as service request/response sections, this property is defined as an empty string. """ return self._source_file_path - @property - def source_file_path_to_root(self) -> Path: - """ - The path to the folder that is the root namespace folder for the `source_file_path` this type was read from. - The `source_file_path` will always be relative to the `source_file_path_to_root` but not all types that share - the same `root_namespace` will have the same path to their root folder since types may be contributed to a - root namespace from several different file trees. For example: - - ``` - path0 = "workspace_0/project_a/types/animal/feline/Tabby.1.0.dsdl" - path1 = "workspace_1/project_b/types/animal/canine/Boxer.1.0.dsdl" - ``` - - In these examples path0 and path1 will produce composite types with `animal` as the root namespace but both - with have different `source_file_path_to_root` paths. - """ - return self._path_to_root_namespace - @property def alignment_requirement(self) -> int: # This is more general than required by the Specification, but it is done this way in case if we decided @@ -721,25 +681,19 @@ def iterate_fields_with_offsets( raise TypeError("Service types do not have serializable fields. Use either request or response.") -# +--[UNIT TESTS]-----------------------------------------------------------------------------------------------------+ - - def _unittest_composite_types() -> None: # pylint: disable=too-many-statements - from typing import Optional - from pytest import raises - + from ._primitive import SignedIntegerType, FloatType from ._array import FixedLengthArrayType, VariableLengthArrayType - from ._primitive import FloatType, SignedIntegerType - def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: + def try_name(name: str) -> CompositeType: return StructureType( name=name, version=Version(0, 1), attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=file_path or Path(*name.split(".")), + source_file_path=Path(), has_parent_service=False, ) @@ -767,9 +721,6 @@ def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: with raises(InvalidNameError, match="(?i).*cannot contain.*"): try_name("namespace.n-s.T") - with raises(InvalidNameError, match=".*Source file directory structure is not consistent.*"): - try_name("a.Foo", Path("foo/bar/b/Foo.0.1.dsdl")) - assert try_name("root.nested.T").full_name == "root.nested.T" assert try_name("root.nested.T").full_namespace == "root.nested" assert try_name("root.nested.T").root_namespace == "root" @@ -782,7 +733,7 @@ def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=Path("a", "A"), + source_file_path=Path(), has_parent_service=False, ) @@ -797,7 +748,7 @@ def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: ], deprecated=False, fixed_port_id=None, - source_file_path=Path("a", "A"), + source_file_path=Path(), has_parent_service=False, ) @@ -811,7 +762,7 @@ def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: ], deprecated=False, fixed_port_id=None, - source_file_path=Path("uavcan", "node", "Heartbeat"), + source_file_path=Path(), has_parent_service=False, ) assert u["a"].name == "a" @@ -837,7 +788,7 @@ def try_name(name: str, file_path: Optional[Path] = None) -> CompositeType: ], deprecated=False, fixed_port_id=None, - source_file_path=Path("a", "A"), + source_file_path=Path(), has_parent_service=False, ) assert s["a"].name == "a" @@ -896,7 +847,7 @@ def try_union_fields(field_types: typing.List[SerializableType]) -> UnionType: attributes=atr, deprecated=False, fixed_port_id=None, - source_file_path=Path("a") / "A", + source_file_path=Path(), has_parent_service=False, ) @@ -940,7 +891,7 @@ def try_union_fields(field_types: typing.List[SerializableType]) -> UnionType: # The reference values for the following test are explained in the array tests above tu8 = UnsignedIntegerType(8, cast_mode=PrimitiveType.CastMode.TRUNCATED) small = VariableLengthArrayType(tu8, 2) - outer = FixedLengthArrayType(small, 2) # un-padded bit length values: {4, 12, 20, 28, 36} + outer = FixedLengthArrayType(small, 2) # unpadded bit length values: {4, 12, 20, 28, 36} # Above plus one bit to each, plus 16-bit for the unsigned integer field assert try_union_fields( @@ -961,7 +912,7 @@ def try_struct_fields(field_types: typing.List[SerializableType]) -> StructureTy attributes=atr, deprecated=False, fixed_port_id=None, - source_file_path=Path("a") / "A", + source_file_path=Path(), has_parent_service=False, ) @@ -993,12 +944,9 @@ def try_struct_fields(field_types: typing.List[SerializableType]) -> StructureTy def _unittest_field_iterators() -> None: # pylint: disable=too-many-locals - import itertools - from pytest import raises - - from ._array import FixedLengthArrayType, VariableLengthArrayType from ._primitive import BooleanType, FloatType + from ._array import FixedLengthArrayType, VariableLengthArrayType saturated = PrimitiveType.CastMode.SATURATED _seq_no = 0 @@ -1012,7 +960,7 @@ def make_type(meta: typing.Type[CompositeType], attributes: typing.Iterable[Attr attributes=attributes, deprecated=False, fixed_port_id=None, - source_file_path=Path("fake_root") / "ns" / f"Type{str(_seq_no)}", + source_file_path=Path(), has_parent_service=False, ) @@ -1280,7 +1228,7 @@ def validate_iterator( attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=Path("ns", "S_1_0.dsdl"), + source_file_path=Path(), has_parent_service=True, ), response=StructureType( @@ -1289,7 +1237,7 @@ def validate_iterator( attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=Path("ns", "S_1_0.dsdl"), + source_file_path=Path(), has_parent_service=True, ), fixed_port_id=None, @@ -1303,7 +1251,7 @@ def validate_iterator( attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=Path("ns", "XX_1_0.dsdl"), + source_file_path=Path(), has_parent_service=True, ), response=StructureType( @@ -1312,31 +1260,8 @@ def validate_iterator( attributes=[], deprecated=True, fixed_port_id=None, - source_file_path=Path("ns", "XX_1_0.dsdl"), - has_parent_service=True, - ), - fixed_port_id=None, - ) - - with raises(ValueError): # Request/response consistency error (internal failure) - ServiceType( - request=StructureType( - name="ns.XX.Request", - version=Version(1, 0), - attributes=[], - deprecated=False, - fixed_port_id=None, - source_file_path=Path("ns", "XX_1_0.dsdl"), - has_parent_service=True, - ), - response=StructureType( - name="ns.XX.Response", - version=Version(1, 0), - attributes=[], - deprecated=False, - fixed_port_id=None, - source_file_path=Path("ns", "YY_1_0.dsdl"), - has_parent_service=True, + source_file_path=Path(), + has_parent_service=False, ), fixed_port_id=None, ) @@ -1348,7 +1273,7 @@ def validate_iterator( attributes=[], deprecated=False, fixed_port_id=None, - source_file_path=Path("e", "E_0_1.dsdl"), + source_file_path=Path(), has_parent_service=False, ) validate_iterator(e, []) diff --git a/pydsdl/_test.py b/pydsdl/_test.py index ca9c373..3e4048f 100644 --- a/pydsdl/_test.py +++ b/pydsdl/_test.py @@ -2,7 +2,6 @@ # This software is distributed under the terms of the MIT License. # Author: Pavel Kirienko -# cSpell: words iceb # pylint: disable=global-statement,protected-access,too-many-statements,consider-using-with,redefined-outer-name import tempfile @@ -63,7 +62,6 @@ def parse_definition( ) -> _serializable.CompositeType: return definition.read( lookup_definitions, - [], print_output_handler=lambda line, text: print("Output from line %d:" % line, text), allow_unregulated_fixed_port_id=False, ) @@ -316,7 +314,7 @@ def _unittest_comments(wrkspc: Workspace) -> None: uint8 CHARACTER = '#' # comment on constant int8 a # comment on field - int8 a_prime + int8 aprime @assert 1 == 1 # toss one in for confusion void2 # comment on padding field saturated int64[<33] b @@ -424,7 +422,7 @@ def _unittest_error(wrkspc: Workspace) -> None: def standalone(rel_path: str, definition: str, allow_unregulated: bool = False) -> _serializable.CompositeType: return wrkspc.parse_new(rel_path, definition + "\n").read( - [], [], lambda *_: None, allow_unregulated + [], lambda *_: None, allow_unregulated ) # pragma: no branch with raises(_error.InvalidDefinitionError, match="(?i).*port ID.*"): @@ -756,20 +754,20 @@ def print_handler(line_number: int, text: str) -> None: wrkspc.parse_new( "ns/A.1.0.dsdl", "# line number 1\n" "# line number 2\n" "@print 2 + 2 == 4 # line number 3\n" "# line number 4\n" "@sealed\n", - ).read([], [], print_handler, False) + ).read([], print_handler, False) assert printed_items assert printed_items[0] == 3 assert printed_items[1] == "true" - wrkspc.parse_new("ns/B.1.0.dsdl", "@print false\n@sealed").read([], [], print_handler, False) + wrkspc.parse_new("ns/B.1.0.dsdl", "@print false\n@sealed").read([], print_handler, False) assert printed_items assert printed_items[0] == 1 assert printed_items[1] == "false" wrkspc.parse_new( "ns/Offset.1.0.dsdl", "@print _offset_ # Not recorded\n" "uint8 a\n" "@print _offset_\n" "@extent 800\n" - ).read([], [], print_handler, False) + ).read([], print_handler, False) assert printed_items assert printed_items[0] == 3 assert printed_items[1] == "{8}" diff --git a/setup.cfg b/setup.cfg index c58e783..8d934ba 100644 --- a/setup.cfg +++ b/setup.cfg @@ -35,7 +35,7 @@ include = pydsdl* # -------------------------------------------------- PYTEST -------------------------------------------------- [tool:pytest] -testpaths = pydsdl test +testpaths = pydsdl norecursedirs = third_party python_files = *.py python_classes = _UnitTest diff --git a/test/test_public_types.py b/test/test_public_types.py deleted file mode 100644 index 8463f97..0000000 --- a/test/test_public_types.py +++ /dev/null @@ -1,47 +0,0 @@ -# Copyright (C) OpenCyphal Development Team -# Copyright Amazon.com Inc. or its affiliates. -# SPDX-License-Identifier: MIT - -# pylint: disable=redefined-outer-name -# pylint: disable=logging-fstring-interpolation -import cProfile -import io -import pstats -from pathlib import Path -from pstats import SortKey - -import pydsdl - - -def _unittest_public_types_namespaces(public_types: Path) -> None: - """ - Sanity check to ensure that the public types can be read. This also allows us to debug - against a real dataset. - """ - pr = cProfile.Profile() - pr.enable() - _ = pydsdl.read_namespace(public_types) - pr.disable() - s = io.StringIO() - sortby = SortKey.TIME - ps = pstats.Stats(pr, stream=s).sort_stats(sortby) - ps.print_stats() - print(s.getvalue()) - - -def _unittest_public_types_files(public_types: Path) -> None: - """ - Sanity check to ensure that the public types can be read. This also allows us to debug - against a real dataset. - """ - pr = cProfile.Profile() - pr.enable() - node_types = list(public_types.glob("node/**/*.dsdl")) - assert len(node_types) > 0 - _ = pydsdl.read_files(node_types, {public_types}) - pr.disable() - s = io.StringIO() - sortby = SortKey.TIME - ps = pstats.Stats(pr, stream=s).sort_stats(sortby) - ps.print_stats() - print(s.getvalue())