diff --git a/.github/workflows/dissect-ci.yml b/.github/workflows/dissect-ci.yml index abba3544..79f7007b 100644 --- a/.github/workflows/dissect-ci.yml +++ b/.github/workflows/dissect-ci.yml @@ -12,6 +12,8 @@ jobs: ci: uses: fox-it/dissect-workflow-templates/.github/workflows/dissect-ci-template.yml@main secrets: inherit + with: + run-benchmarks: true publish: if: ${{ github.ref_name == 'main' || github.ref_type == 'tag' }} diff --git a/README.md b/README.md index f82ea289..52b2750c 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,59 @@ $ rdump output.records.gz ``` +### Declarative record definitions + +Besides the `RecordDescriptor` API shown above, records can also be defined declaratively by subclassing `RecordBase` +and annotating fields, like a `dataclass` or `NamedTuple`: + +```python +from datetime import datetime +from typing import Annotated + +from flow.record.declarative import RecordBase, field + + +class HttpRequestRecord(RecordBase, name="http/request"): + ts: datetime + url: str # native types map (e.g. str -> string) + status: Annotated[int, "uint32"] # or field(typename="uint32") + remote: str + + +record = HttpRequestRecord(ts=datetime.now(), url="http://flow.record", status=200, remote="127.0.0.1") +``` + +Inheritance works as expected — a subclass extends its parent's fields: + +```python +class HttpResponseRecord(HttpRequestRecord, name="http/response"): + body: bytes +``` + +Use `field(init=False)` for derived fields and populate them from `InitVar` inputs in a `__post_init__` hook: + +```python +from dataclasses import InitVar + + +class HostRecord(RecordBase, name="example/host"): + hostname: str = field(init=False) + target: InitVar[object] + + def __post_init__(self, target: object) -> None: + self.hostname = target.hostname +``` + +> [!NOTE] +> Field annotations are evaluated at runtime (via `get_type_hints`). If your project uses Ruff's `flake8-type-checking` +> (`TC`) rules, add `RecordBase` to the runtime-evaluated base classes once, so imports used only in field annotations +> are not moved into a `TYPE_CHECKING` block: +> +> ```toml +> [tool.ruff.lint.flake8-type-checking] +> runtime-evaluated-base-classes = ["flow.record.declarative.RecordBase"] +> ``` + ### Selectors We can also use `selectors` for filtering and selecting records using a query (Python like syntax), e.g.: diff --git a/flow/record/base.py b/flow/record/base.py index b64e5a3f..8e261388 100644 --- a/flow/record/base.py +++ b/flow/record/base.py @@ -114,6 +114,7 @@ def _unpack(__cls, {args}): {unpack_code} """ +_FIELDTYPES_PREFIX = "flow.record.fieldtypes" if env_excluded_fields := os.environ.get("FLOW_RECORD_IGNORE"): IGNORE_FIELDS_FOR_COMPARISON = set(env_excluded_fields.split(",")) @@ -153,6 +154,11 @@ def _unpack(cls, data: Any) -> Any: return data +def _default_is_trivial(field_type: type[FieldType]) -> bool: + """Return whether ``field_type`` uses the base :meth:`FieldType.default` (i.e. returns ``None``).""" + return field_type.default.__func__ is FieldType.default.__func__ + + class Record: __slots__ = () @@ -441,10 +447,7 @@ def _generate_record_class(name: str, fields: tuple[tuple[str, str]]) -> type: args = ", ".join([f"{k}=None" for k in all_fields]) unpack_code = "\t\treturn __cls(\n" for field in all_fields.values(): - if field.type.default == FieldType.default: - default = FieldType.default() - else: - default = f"_field_{field.name}.type.default()" + default = FieldType.default() if _default_is_trivial(field.type) else f"_field_{field.name}.type.default()" init_code += f"\t\t__self.{field.name} = {field.name} if {field.name} is not None else {default}\n" unpack_code += ( "\t\t\t{field} = _field_{field}.type._unpack({field}) " + "if {field} is not None else {default},\n" @@ -949,8 +952,6 @@ def fieldtype(clspath: str) -> FieldType: Returns: The FieldType class. """ - base_module_path = "flow.record.fieldtypes" - if clspath.endswith("[]"): origpath = clspath clspath = clspath[:-2] @@ -962,13 +963,13 @@ def fieldtype(clspath: str) -> FieldType: raise AttributeError(f"Invalid field type: {clspath}") namespace, _, clsname = clspath.rpartition(".") - module_path = f"{base_module_path}.{namespace}" if namespace else base_module_path + module_path = f"{_FIELDTYPES_PREFIX}.{namespace}" if namespace else _FIELDTYPES_PREFIX mod = importlib.import_module(module_path) fieldtype_cls = getattr(mod, clsname) if islist: - base_mod = importlib.import_module(base_module_path) + base_mod = importlib.import_module(_FIELDTYPES_PREFIX) listtype = type(origpath, base_mod.typedlist.__bases__, dict(base_mod.typedlist.__dict__)) listtype.__type__, fieldtype_cls = fieldtype_cls, listtype diff --git a/flow/record/declarative.py b/flow/record/declarative.py new file mode 100644 index 00000000..395aed13 --- /dev/null +++ b/flow/record/declarative.py @@ -0,0 +1,446 @@ +"""Declarative (subclass) syntax for defining records. + +Besides the :class:`~flow.record.RecordDescriptor` API, records can also be +defined declaratively by subclassing :class:`RecordBase` and annotating fields, +like a ``dataclass`` or ``NamedTuple``. The metaclass builds the +:class:`~flow.record.RecordDescriptor` and folds the record behaviour into the +class, so instances are genuine ``Record`` objects while type checkers see a +typed constructor and attribute access (via :func:`typing.dataclass_transform`):: + + from datetime import datetime + from typing import Annotated + + from flow.record.declarative import RecordBase, field + + + class HttpRequest(RecordBase, name="http/request"): + ts: datetime + url: str # native types map (e.g. str -> string) + status: Annotated[int, "uint32"] # or field(typename="uint32") + remote: str + + + record = HttpRequest(ts=datetime.now(), url="http://flow.record", status=200, remote="127.0.0.1") + +Inheritance works as expected -- a subclass extends its parent's fields:: + + class HttpResponse(HttpRequest, name="http/response"): + body: bytes + +Use ``field(init=False)`` for derived fields and populate them from ``InitVar`` +inputs in a ``__post_init__`` hook:: + + from dataclasses import InitVar + + + class Host(RecordBase, name="example/host"): + hostname: str = field(init=False) + target: InitVar[object] + + def __post_init__(self, target: object) -> None: + self.hostname = target.hostname + +Note: field annotations are evaluated at runtime (via :func:`typing.get_type_hints`). +Projects using Ruff's ``flake8-type-checking`` (``TC``) rules should add +``RecordBase`` to the runtime-evaluated base classes once, so that imports used +only in field annotations are not moved into a ``TYPE_CHECKING`` block:: + + [tool.ruff.lint.flake8-type-checking] + runtime-evaluated-base-classes = ["flow.record.declarative.RecordBase"] +""" + +from __future__ import annotations + +import ast +import dataclasses +import inspect +import sys +import textwrap +from datetime import datetime +from pathlib import Path +from typing import TYPE_CHECKING, Annotated, Any, ClassVar, NamedTuple, get_args, get_origin, get_type_hints + +from flow.record.base import ( + _FIELDTYPES_PREFIX, + RECORD_VERSION, + FieldType, + Record, + RecordDescriptor, + RecordField, + _default_is_trivial, + _utcnow, +) +from flow.record.whitelist import WHITELIST + +try: + from typing import dataclass_transform # novermin # Python 3.11+ +except ImportError: # Python 3.10 + try: + from typing_extensions import dataclass_transform + except ImportError: + # dataclass_transform is a typing-only marker with no runtime effect, so a + # no-op keeps imports working without a typing_extensions runtime dependency. + def dataclass_transform(**kwargs: Any) -> Callable[[Any], Any]: + return lambda obj: obj + + +if TYPE_CHECKING: + from collections.abc import Callable + +PY_311_OR_HIGHER = sys.version_info >= (3, 11) + +# Native Python type -> flow.record typename +# Use `field(typename=...)` or `Annotated[int, "..."]` for a precise type +_PY_TYPE_TO_TYPENAME: dict[type, str] = { + str: "string", + int: "varint", + float: "float", + bool: "boolean", + bytes: "bytes", + datetime: "datetime", + Path: "path", +} + +# Metadata key under which `field(typename=...)` stashes an explicit typename +_TYPENAME_KEY = "flow.record.typename" + + +def field( + *, + default: Any = dataclasses.MISSING, + default_factory: Any = dataclasses.MISSING, + init: bool = True, + typename: str | None = None, +) -> Any: + """Field specifier for declarative records, wrapping :func:`dataclasses.field`. + + Use in place of a bare annotation to set a ``default``/``default_factory``, + mark a derived field with ``init=False``, or pin a ``typename``. + """ + metadata = {_TYPENAME_KEY: typename} if typename is not None else {} + kwargs: dict[str, Any] = {"init": init, "metadata": metadata} + if default is not dataclasses.MISSING: + kwargs["default"] = default + if default_factory is not dataclasses.MISSING: + kwargs["default_factory"] = default_factory + return dataclasses.field(**kwargs) + + +def _resolve_typename(annotation: Any) -> str: + """Map a resolved field annotation to a ``flow.record`` typename.""" + if get_origin(annotation) is Annotated: + base, *metadata = get_args(annotation) + for meta in metadata: + if isinstance(meta, str): + return meta + annotation = base + + if isinstance(annotation, type) and issubclass(annotation, FieldType): + return _fieldtype_typename(annotation) + + if isinstance(annotation, type) and annotation in _PY_TYPE_TO_TYPENAME: + return _PY_TYPE_TO_TYPENAME[annotation] + + raise TypeError(f"Failed to map annotation {annotation!r} to a flow.record fieldtype") + + +def _fieldtype_typename(annotation: type[FieldType]) -> str: + """Resolve a ``flow.record`` typename from a fieldtype class. + + Top-level types map directly (``uint32`` -> ``"uint32"``), namespaced types + like :class:`~flow.record.fieldtypes.net.ipaddress` are matched against the + whitelist by progressively shortening the module path, so classes that are + re-exported one level up from where they are defined still resolve + (``net.ip.ipaddress`` -> ``net.ipaddress``). + """ + name = annotation.__name__ + module = annotation.__module__ + parts = module[len(_FIELDTYPES_PREFIX) + 1 :].split(".") if module.startswith(_FIELDTYPES_PREFIX + ".") else [] + + # Try the most specific namespace first, then shorten towards the bare name + for i in range(len(parts), -1, -1): + namespace = ".".join(parts[:i]) + candidate = f"{namespace}.{name}" if namespace else name + if candidate in WHITELIST: + return candidate + + # Fall back to the bare class name, RecordDescriptor will validate the typename + return name + + +def _is_classvar(hint: Any) -> bool: + return hint is ClassVar or get_origin(hint) is ClassVar + + +def _is_initvar(hint: Any) -> bool: + if isinstance(hint, str): + text = hint.strip().removeprefix("dataclasses.") + return text == "InitVar" or text.startswith("InitVar[") + return hint is dataclasses.InitVar or type(hint) is dataclasses.InitVar + + +def _resolve_hints(cls: type) -> tuple[dict[str, Any], list[str]]: + """Resolve merged field annotations, split into fields and init-only var names.""" + if PY_311_OR_HIGHER: + fields: dict[str, Any] = {} + initvars: list[str] = [] + + for name, hint in get_type_hints(cls, include_extras=True).items(): + if _is_initvar(hint): + initvars.append(name) + else: + fields[name] = hint + + return fields, initvars + + # Python 3.10 fallback: hide InitVar annotations from get_type_hints. + initvars = [] + removed: list[tuple[dict[str, Any], str, Any]] = [] + for klass in reversed(cls.__mro__): + if not (ann := klass.__dict__.get("__annotations__")): + continue + + for name in list(ann): + if _is_initvar(ann[name]): + if name not in initvars: + initvars.append(name) + + removed.append((ann, name, ann[name])) + del ann[name] + + try: + hints = get_type_hints(cls, include_extras=True) + finally: + for ann, name, value in removed: + ann[name] = value + + return hints, initvars + + +# Sentinel marking an init-only var that was not supplied to the constructor +_MISSING = object() + + +def _generate_init( + init_field_names: list[str], + derived_field_names: list[str], + initvar_names: list[str], + field_types: dict[str, Any], + field_specs: dict[str, dataclasses.Field], + has_post_init: bool, +) -> Callable[..., None]: + """Build a specialised ``__init__`` for a declarative record class. + + The generated body is straight-line code with field defaults inlined. + Assignments go through ``Record.__setattr__`` so values are still coerced to their fieldtypes. + """ + if initvar_names and not has_post_init: + raise TypeError(f"init-only vars {sorted(initvar_names)} require a __post_init__ hook") + + global_ns: dict[str, Any] = { + "__utcnow": _utcnow, + "__RECORD_VERSION": RECORD_VERSION, + "__MISSING": _MISSING, + } + + def default_expr(name: str) -> str | None: + """Inline expression for a field's default, or ``None`` to use ``None``. + + An explicit ``field(default=...)``/``default_factory=...`` wins, otherwise + the fieldtype's own default is used (``None`` for most types). + """ + spec = field_specs.get(name) + if spec is not None and spec.default is not dataclasses.MISSING: + global_ns[gname := f"__default_{name}"] = spec.default + return gname + + if spec is not None and spec.default_factory is not dataclasses.MISSING: + global_ns[gname := f"__default_{name}"] = spec.default_factory + return f"{gname}()" + + if _default_is_trivial(field_type := field_types[name]): + return None + + global_ns[gname := f"__default_{name}"] = field_type.default + return f"{gname}()" + + params = [f"{name}=None" for name in init_field_names] + params += [f"{name}=__MISSING" for name in initvar_names] + params += ["_source=None", "_classification=None", "_generated=None"] + + lines = [f"def __init__(__self, {', '.join(params)}):"] + + for name in init_field_names: + if (expr := default_expr(name)) is None: + lines.append(f" __self.{name} = {name}") + else: + lines.append(f" __self.{name} = {name} if {name} is not None else {expr}") + + for name in derived_field_names: + expr = default_expr(name) + lines.append(f" __self.{name} = {expr or 'None'}") + + lines.append(" __self._source = _source") + lines.append(" __self._classification = _classification") + + if initvar_names: + lines.append(" __initvars = {}") + for name in initvar_names: + lines.append(f" if {name} is not __MISSING:") + lines.append(f" __initvars[{name!r}] = {name}") + lines.append(" __self.__post_init__(**__initvars)") + elif has_post_init: + lines.append(" __self.__post_init__()") + + lines.append(" __self._generated = _generated or __utcnow()") + lines.append(" __self._version = __RECORD_VERSION") + + local_ns: dict[str, Any] = {} + exec("\n".join(lines), global_ns, local_ns) + return local_ns["__init__"] + + +@dataclass_transform( + eq_default=True, + kw_only_default=False, + field_specifiers=(field, dataclasses.field, dataclasses.Field), +) +class _RecordMeta(type): + """Metaclass that turns an annotated subclass into a record class.""" + + if TYPE_CHECKING: + __descriptor__: ClassVar[RecordDescriptor] + + def __new__(mcs, cls_name: str, bases: tuple[type, ...], namespace: dict[str, Any], **kwargs: Any): + cls = super().__new__(mcs, cls_name, bases, namespace) + + # The base sentinel itself (no RecordBase ancestor): nothing to describe + if not any(isinstance(b, _RecordMeta) for b in bases): + return cls + + record_name = kwargs.get("name") or namespace.get("__record_name__", cls_name) + + hints, initvars = _resolve_hints(cls) + + field_tuples: list[tuple[str, str]] = [] + init_field_names: list[str] = [] + field_specs: dict[str, dataclasses.Field] = {} + for fname, hint in hints.items(): + if fname.startswith("__") or _is_classvar(hint): + continue + + spec = getattr(cls, fname, None) + if is_field_spec := isinstance(spec, dataclasses.Field): + field_specs[fname] = spec + + # Explicit `field(typename=...)` wins, else infer from the annotation + typename = spec.metadata.get(_TYPENAME_KEY) if is_field_spec else None + field_tuples.append((typename or _resolve_typename(hint), fname)) + + # `field(init=False)` marks a derived field, not a constructor arg + if not (is_field_spec and spec.init is False): + init_field_names.append(fname) + + descriptor = RecordDescriptor(record_name, field_tuples) + field_types: dict[str, Any] = {name: RecordField(name, typename).type for typename, name in field_tuples} + for rname, rfield in RecordDescriptor.get_required_fields().items(): + field_types[rname] = rfield.type + + # Derived (`init=False`) fields are initialised from their default and + # then populated by `__post_init__`. Keep them out of the constructor + init_set = set(init_field_names) + derived_field_names = [name for _, name in field_tuples if name not in init_set] + # Resolve the `__post_init__` hook once so the generated `__init__` + # doesn't probe for it on every construction + has_post_init = callable(getattr(cls, "__post_init__", None)) + + # `__slots__` is set as a plain attribute (the field-name list that + # Record internals iterate), NOT a real slots declaration, so records + # stay dict-based and multi-level inheritance is layout-conflict free + record_cls: Any = cls + record_cls._desc = descriptor + record_cls._field_types = field_types + record_cls.__slots__ = tuple(field_types.keys()) + record_cls.__descriptor__ = descriptor + # Compile a constructor for this exact field set + record_cls.__init__ = _generate_init( + init_field_names, derived_field_names, initvars, field_types, field_specs, has_post_init + ) + return cls + + def __init__(cls, cls_name: str, bases: tuple[type, ...], namespace: dict[str, Any], **kwargs: Any) -> None: + # Swallow the `name=` (and any other) class keyword so `type.__init__` doesn't choke + super().__init__(cls_name, bases, namespace) + + +class RecordBase(Record, metaclass=_RecordMeta): + """Base class for declarative record definitions. + + Subclass and annotate fields. Set the record type name with ``name="cat/type"`` + (class keyword) or ``__record_name__``, defaulting to the class name. + + Declare init-only inputs with ``from dataclasses import InitVar`` and populate + derived fields from them in ``__post_init__``. + """ + + if TYPE_CHECKING: + __descriptor__: ClassVar[RecordDescriptor] + + +class FieldInfo(NamedTuple): + """Type and optional docstring information for a declarative record field.""" + + typename: str + doc: str | None + + +def get_field_info(cls: type) -> dict[str, FieldInfo]: + """Return a ``{field_name: FieldInfo}`` mapping for a record class. + + The ``typename`` always comes from the record descriptor. The ``doc`` is the + PEP 224-style attribute docstring (a string literal directly following a field + annotation), recovered from the class source. It is ``None`` when a field has + no docstring or when the source cannot be read. + """ + # Declarative records expose `__descriptor__`, classic record types expose `_desc`. + descriptor: RecordDescriptor | None = getattr(cls, "__descriptor__", None) or getattr(cls, "_desc", None) + if descriptor is None: + raise TypeError(f"{cls!r} is not a record class") + + docstrings = _extract_field_docstrings(cls) + return {name: FieldInfo(typename, docstrings.get(name)) for typename, name in descriptor.get_field_tuples()} + + +def _extract_field_docstrings(cls: type) -> dict[str, str]: + """Best-effort attribute-docstring extraction across a record's MRO. + + Walks base-to-derived so more-derived docstrings win, and silently skips any + class whose source is unavailable or cannot be parsed. + """ + docstrings: dict[str, str] = {} + + for klass in reversed(cls.__mro__): + if "__descriptor__" not in vars(klass): + continue + + try: + (class_def,) = ast.parse(textwrap.dedent(inspect.getsource(klass))).body + except (OSError, TypeError, SyntaxError, ValueError): + continue + + if not isinstance(class_def, ast.ClassDef): + continue + + # An attribute docstring is a bare string literal directly following the + # field's annotation, so pair each node with the one after it + for annotation, following in zip(class_def.body, class_def.body[1:], strict=False): + if ( + isinstance(annotation, ast.AnnAssign) + and isinstance(annotation.target, ast.Name) + and isinstance(following, ast.Expr) + and isinstance(following.value, ast.Constant) + and isinstance(following.value.value, str) + ): + docstrings[annotation.target.id] = following.value.value + + return docstrings diff --git a/pyproject.toml b/pyproject.toml index b6b1466b..e7b30a40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,6 +110,7 @@ build = [ lint = [ "ruff==0.13.1", "vermin", + "typing_extensions", ] dev = [ {include-group = "test"}, @@ -181,6 +182,7 @@ convention = "google" [tool.ruff.lint.flake8-type-checking] strict = true +runtime-evaluated-base-classes = ["flow.record.declarative.RecordBase"] [tool.ruff.lint.per-file-ignores] "tests/_docs/**" = ["INP001"] @@ -194,4 +196,3 @@ include = ["flow.*"] [tool.setuptools_scm] version_file = "flow/record/version.py" - diff --git a/tests/conftest.py b/tests/conftest.py index 86628485..57fb3053 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,26 @@ from __future__ import annotations +import importlib.util import typing import pytest from flow.record.context import APP_CONTEXT +HAS_BENCHMARK = importlib.util.find_spec("pytest_benchmark") is not None + + +def pytest_configure(config: pytest.Config) -> None: + if not HAS_BENCHMARK: + # If we don't have pytest-benchmark (or pytest-codspeed) installed, register the benchmark marker ourselves + # to avoid pytest warnings + config.addinivalue_line("markers", "benchmark: mark test for benchmarking (requires pytest-benchmark)") + + +def pytest_runtest_setup(item: pytest.Item) -> None: + if not HAS_BENCHMARK and item.get_closest_marker("benchmark") is not None: + pytest.skip("pytest-benchmark is not installed") + @pytest.fixture(autouse=True) def reset_app_context() -> typing.Generator[None, None, None]: diff --git a/tests/record/test_declarative.py b/tests/record/test_declarative.py new file mode 100644 index 00000000..b9fc14ce --- /dev/null +++ b/tests/record/test_declarative.py @@ -0,0 +1,271 @@ +from __future__ import annotations + +import inspect +import io +import re +from dataclasses import InitVar +from datetime import datetime, timezone +from typing import Annotated, Any + +import pytest + +from flow.record import RecordDescriptor, RecordReader, RecordWriter +from flow.record.base import Record +from flow.record.declarative import FieldInfo, RecordBase, field, get_field_info +from flow.record.fieldtypes import net, string, uint32 + + +def test_declarative_basic() -> None: + class TestRecord(RecordBase, name="test/record"): + url: str + status: Annotated[int, "uint32"] + + r = TestRecord(url="http://flow.record", status=200) + + assert isinstance(r, Record) + assert isinstance(r, TestRecord) + assert r.url == "http://flow.record" + assert r.status == 200 + + # Values are coerced to their fieldtypes. + assert isinstance(r.url, string) + assert isinstance(r.status, uint32) + + +def test_declarative_name() -> None: + class KwargName(RecordBase, name="test/kwarg"): + a: int + + class AttrName(RecordBase): + __record_name__ = "test/attr" + a: int + + class DefaultName(RecordBase): + a: int + + assert KwargName.__descriptor__.name == "test/kwarg" + assert AttrName.__descriptor__.name == "test/attr" + assert DefaultName.__descriptor__.name == "DefaultName" + + +def test_declarative_typenames() -> None: + class TestRecord(RecordBase, name="test/types"): + a: str + b: int + c: uint32 + d: Annotated[int, "uint16"] + e: int = field(typename="uint32") + + assert TestRecord.__descriptor__.get_field_tuples() == ( + ("string", "a"), + ("varint", "b"), + ("uint32", "c"), + ("uint16", "d"), + ("uint32", "e"), + ) + + +def test_declarative_fieldtype_class_annotation() -> None: + class TestRecord(RecordBase, name="test/fieldtype_class"): + count: uint32 # top-level fieldtype class + ip: net.ipaddress # namespaced fieldtype class + + # Namespaced fieldtypes resolve to their whitelisted typename + assert TestRecord.__descriptor__.get_field_tuples() == ( + ("uint32", "count"), + ("net.ipaddress", "ip"), + ) + + r = TestRecord(count=5, ip="1.1.1.1") + assert isinstance(r.count, uint32) + assert isinstance(r.ip, net.ipaddress) + assert str(r.ip) == "1.1.1.1" + + +def test_declarative_positional() -> None: + class TestRecord(RecordBase, name="test/positional"): + a: int + b: str + + r = TestRecord(1, "two") + assert r.a == 1 + assert r.b == "two" + + +def test_declarative_defaults() -> None: + class TestRecord(RecordBase, name="test/default"): + a: int = field(default=42) + b: str = field(default_factory=lambda: "generated") + + # `field(default=...)`/`default_factory=...` are applied when no value is given. + r = TestRecord() + assert r.a == 42 + assert r.b == "generated" + + # Explicit values still win over the default. + assert TestRecord(a=1).a == 1 + + +def test_declarative_unexpected_keyword() -> None: + class TestRecord(RecordBase, name="test/unexpected"): + a: int + + with pytest.raises(TypeError): + TestRecord(a=1, nope=2) + + +def test_declarative_inheritance() -> None: + class Base(RecordBase, name="test/base"): + a: int + b: str + + class Child(Base, name="test/child"): + c: int + + r = Child(a=1, b="two", c=3) + + assert isinstance(r, Base) + assert isinstance(r, Child) + # Fields are merged across the MRO, base fields first. + assert Child.__descriptor__.get_field_tuples() == ( + ("varint", "a"), + ("string", "b"), + ("varint", "c"), + ) + assert (r.a, r.b, r.c) == (1, "two", 3) + + +def test_declarative_post_init() -> None: + class Target: + hostname = "host01" + domain = "example.com" + + class Base(RecordBase, name="test/target"): + hostname: str = field(init=False) + domain: str = field(init=False) + _target: InitVar[Any] + + def __post_init__(self, _target: Any) -> None: + self.hostname = _target.hostname + self.domain = _target.domain + + class Child(Base, name="test/target_child"): + a: int + + r = Child(a=1, _target=Target()) + + assert r.hostname == "host01" + assert r.domain == "example.com" + assert r.a == 1 + # Init-only vars are not stored. + assert "_target" not in r._asdict() + + +def test_declarative_init_false_excluded_from_constructor() -> None: + class TestRecord(RecordBase, name="test/init_false"): + a: int + b: str = field(init=False) + + # `a` is a constructor argument; `b` (init=False) is not part of the signature + # and is initialised to its default instead. + r = TestRecord(a=1) + assert r.a == 1 + assert r.b is None + + # Only init fields count towards positional binding. + assert TestRecord(2).a == 2 + + # The derived field is rejected as a constructor argument. + with pytest.raises(TypeError): + TestRecord(a=1, b="nope") + + +def test_declarative_initvar_without_post_init() -> None: + # An init-only var with no __post_init__ to consume it is a definition error. + with pytest.raises(TypeError, match=re.escape("init-only vars ['_target'] require a __post_init__ hook")): + + class TestRecord(RecordBase, name="test/no_post_init"): + _target: InitVar[Any] + + +def test_declarative_roundtrip() -> None: + class TestRecord(RecordBase, name="test/roundtrip"): + ts: datetime + url: str + status: int = field(typename="uint32") + + r = TestRecord(ts=datetime.now(timezone.utc), url="http://flow.record", status=200) + + buf = io.BytesIO() + writer = RecordWriter(fileobj=buf) + writer.write(r) + writer.flush() + buf.seek(0) + + [read] = list(RecordReader(fileobj=buf)) + assert read.ts == r.ts + assert read.url == r.url + assert read.status == r.status + assert read._desc.name == "test/roundtrip" + + +def test_declarative_field_docstring() -> None: + class TestRecord(RecordBase, name="test/doc"): + url: str + """The requested URL.""" + status: int + + # An inline field docstring must not interfere with the record definition. + r = TestRecord(url="http://flow.record", status=200) + assert r.url == "http://flow.record" + + assert get_field_info(TestRecord) == { + "url": FieldInfo("string", "The requested URL."), + "status": FieldInfo("varint", None), + } + + +def test_declarative_field_docstring_inherited() -> None: + class Base(RecordBase, name="test/doc_base"): + a: int + """Field a.""" + + class Child(Base, name="test/doc_child"): + b: str + """Field b.""" + + docs = get_field_info(Child) + # Docstrings are gathered across the whole MRO. + assert docs["a"] == FieldInfo("varint", "Field a.") + assert docs["b"] == FieldInfo("string", "Field b.") + + +def test_declarative_field_docstring_source_unavailable(monkeypatch: pytest.MonkeyPatch) -> None: + class TestRecord(RecordBase, name="test/doc_nosrc"): + url: str + """The requested URL.""" + + def _raise(_: object) -> str: + raise OSError("source not available") + + monkeypatch.setattr(inspect, "getsource", _raise) + + # Typenames still resolve; docstrings degrade to None instead of raising. + assert get_field_info(TestRecord) == {"url": FieldInfo("string", None)} + + +def test_declarative_field_info_classic_record() -> None: + # A classically generated record type has no retrievable source (it is exec'd) + # and no attribute docstrings; get_field_info must still resolve typenames and + # degrade docstrings to None without blowing up. + classic = RecordDescriptor("test/classic", [("string", "url"), ("uint32", "status")]).recordType + + assert get_field_info(classic) == { + "url": FieldInfo("string", None), + "status": FieldInfo("uint32", None), + } + + +def test_declarative_field_docs_not_a_record() -> None: + with pytest.raises(TypeError): + get_field_info(int) diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py new file mode 100644 index 00000000..65108ce0 --- /dev/null +++ b/tests/test_benchmark.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import TYPE_CHECKING, Annotated + +import pytest + +from flow.record import RecordDescriptor +from flow.record.declarative import RecordBase + +if TYPE_CHECKING: + from pytest_benchmark.fixture import BenchmarkFixture + + +ClassicRecord = RecordDescriptor( + "test/benchmark", + [ + ("datetime", "ts"), + ("string", "url"), + ("uint32", "status"), + ("string", "remote"), + ], +) + + +class DeclarativeRecord(RecordBase, name="test/benchmark"): + ts: datetime + url: str + status: Annotated[int, "uint32"] + remote: str + + +@pytest.mark.benchmark +def test_benchmark_classic_init(benchmark: BenchmarkFixture) -> None: + """Benchmark constructing a classic ``RecordDescriptor``-generated record.""" + ts = datetime.now(timezone.utc) + benchmark(lambda: ClassicRecord(ts=ts, url="http://flow.record", status=200, remote="127.0.0.1")) + + +@pytest.mark.benchmark +def test_benchmark_declarative_init(benchmark: BenchmarkFixture) -> None: + """Benchmark constructing a declarative ``RecordBase`` record with the same fields.""" + ts = datetime.now(timezone.utc) + benchmark(lambda: DeclarativeRecord(ts=ts, url="http://flow.record", status=200, remote="127.0.0.1")) diff --git a/tox.ini b/tox.ini index 72ab0dd0..12137c39 100644 --- a/tox.ini +++ b/tox.ini @@ -13,16 +13,25 @@ requires = virtualenv>=20.24.6 [testenv] deps = - pytest pytest-cov coverage dependency_groups = test commands = # Capturing output will fail on pypy, possibly due to this issue: https://github.com/pytest-dev/pytest/issues/5502 - pytest --basetemp="{envtmpdir}" {posargs:--color=yes --capture=no --cov=flow --cov-report=term-missing -v tests} + pytest --basetemp="{envtmpdir}" --import-mode="append" {posargs:--color=yes --capture=no --cov=flow --cov-report=term-missing -v tests} coverage report coverage xml +[testenv:benchmark] +deps = + pytest-benchmark + pytest-codspeed +dependency_groups = test +passenv = + CODSPEED_ENV +commands = + pytest --basetemp="{envtmpdir}" --import-mode="append" -m benchmark {posargs:--color=yes -v tests} + [testenv:build] package = skip dependency_groups = build