Crash - Cannot import name 'FixtureDef' from 'pytest'

[coffeegist]

Bug description

When parsing the following a.py:

"""Data source abstractions for BOFHound parsing pipeline."""

import sys
from abc import ABC, abstractmethod
import logging
from typing import Iterator, List, AsyncIterator
from typing_extensions import override
import glob
import os
from pathlib import Path
from mythic import mythic
from syncer import sync
import base64

from bofhound.logger import logger



class DataSource(ABC):
    """Abstract base class for data sources that provide lines to parse."""

    @abstractmethod
    def get_data_streams(self) -> Iterator['DataStream']:
        """Return an iterator of data streams to process."""
        pass


class DataStream(ABC):
    """Abstract base class representing a single stream of data to parse."""

    @property
    @abstractmethod
    def identifier(self) -> str:
        """Unique identifier for this data stream (e.g., filename, callback ID)."""
        pass

    @abstractmethod
    def lines(self) -> Iterator[str]:
        """Return an iterator of lines from this data stream."""
        pass

    def __str__(self) -> str:
        return self.identifier

class FileDataSource(DataSource):
    """Data source that reads from local files."""

    def __init__(self, input_path: str, filename_pattern: str = "*.log"):
        self.input_path = input_path
        self.filename_pattern = filename_pattern

    def get_data_streams(self) -> Iterator['FileDataStream']:
        """Get file-based data streams."""
        if os.path.isfile(self.input_path):
            yield FileDataStream(self.input_path)
        elif os.path.isdir(self.input_path):
            pattern = f"{self.input_path}/**/{self.filename_pattern}"
            files = glob.glob(pattern, recursive=True)
            files.sort(key=os.path.getmtime)

            for file_path in files:
                yield FileDataStream(file_path)
        else:
            raise ValueError(f"Input path does not exist: {self.input_path}")


class FileDataStream(DataStream):
    """Data stream that reads from a local file."""

    def __init__(self, file_path: str):
        self.file_path = file_path

    @property
    def identifier(self) -> str:
        return self.file_path

    def lines(self) -> Iterator[str]:
        """Read lines from the file."""
        with open(self.file_path, 'r', encoding='utf-8') as f:
            for line in f:
                yield line.rstrip('\n\r')


class MythicCallback:
    """
    Quick and dirty class to hold Mythic callback information
    and allow print statments from the main logic to still work
    """
    def __init__(self, callback, mythic_instance=None):
        self.callback_id = callback["id"]
        self.display_id = callback["display_id"]
        self.domain = callback["domain"]
        self.user = callback["user"]
        self.host = callback["host"]
        self.uuid = callback["agent_callback_id"]
        self._mythic_instance = mythic_instance

    def __repr__(self):
        return f"Mythic callback {self.callback_id} [{self.uuid}]"


class MythicDataSource(DataSource):
    """Data source that fetches data from Mythic server."""

    def __init__(self, mythic_server: str, mythic_token: str):
        self.mythic_server = mythic_server
        self.mythic_token = mythic_token
        self._mythic_instance = None

    def _connect(self):
        logger.debug("Logging into Mythic...")
        try:
            self._mythic_instance = sync(mythic.login(
                apitoken=self.mythic_token,
                server_ip=self.mythic_server,
                server_port=7443,
                timeout=-1,
                logging_level=logging.CRITICAL,
            ))
        except Exception as e:
            logger.error("Error logging into Mythic")
            logger.error(e)
            sys.exit(-1)

        logger.debug("Logged into Mythic successfully")


    def _get_callbacks(self) -> Iterator[MythicCallback]:
        logger.debug("Retrieving callbacks from Mythic...")
        return_attributes = [
            "id",
            "display_id",
            "domain",
            "user",
            "host",
            "agent_callback_id"
        ]

        try:
            if not self._mythic_instance:
                self._connect()

            raw_callbacks = sync(mythic.get_all_callbacks(
                self._mythic_instance,
                custom_return_attributes=",".join(return_attributes)
            ))

            for callback in raw_callbacks:
                yield MythicCallback(callback, self._mythic_instance)

        except Exception as e:
            logger.error("Error retrieving callbacks from Mythic")
            logger.error(e)
            sys.exit(-1)

    @override
    def get_data_streams(self) -> Iterator['MythicDataStream']:
        """
        Get Mythic callback data streams.
        For mythic, instead of processing individual log "files"
        we will processes the taskings given to individual callbacks
        """
        for callback in self._get_callbacks():
            yield MythicDataStream(callback)

class MythicDataStream(DataStream):
    """Data stream that reads from a Mythic callback's task outputs."""

    def __init__(self, callback: MythicCallback):
        self.callback = callback

    @property
    def identifier(self) -> str:
        return f"mythic_callback_{self.callback.callback_id}"

    def lines(self) -> Iterator[str]:
        """Get lines from Mythic callback task outputs."""
        # Get all tasks for this callback
        tasks = self._get_tasks()

        for task in tasks:
            # Get task output
            outputs = self._get_task_output(task["display_id"])

            for output in outputs:
                # Decode and yield each line
                try:
                    decoded_data = base64.b64decode(output["response_text"]).decode("utf-8")
                    for line in decoded_data.splitlines():
                        if line.strip():  # Skip empty lines
                            yield line
                except Exception:
                    continue  # Skip malformed responses

    def _get_tasks(self):
        """Get tasks for the callback."""
        return sync(mythic.get_all_tasks(
            self.callback._mythic_instance,
            callback_display_id=self.callback.display_id
        ))

    def _get_task_output(self, task_id):
        """Get output for a specific task."""
        return sync(mythic.get_all_task_output_by_id(
            self.callback._mythic_instance,
            task_id
        ))

Command used

pylint a.py

Pylint output

pylint crashed with a ``AstroidError`` and with the following stacktrace:

Traceback (most recent call last):
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/lint/pylinter.py", line 788, in _lint_file
    check_astroid_module(module)
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/lint/pylinter.py", line 1020, in check_astroid_module
    retval = self._check_astroid_module(
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/lint/pylinter.py", line 1072, in _check_astroid_module
    walker.walk(node)
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/utils/ast_walker.py", line 87, in walk
    callback(astroid)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pylint_pytest/checkers/fixture.py", line 129, in visit_module
    ret = pytest.main(
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/config/__init__.py", line 150, in main
    arguments directly from the process command line (:data:`sys.argv`).
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/config/__init__.py", line 331, in _prepareconfig
    elif not isinstance(args, list):
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 43, in run_old_style_hookwrapper
    teardown.send(result)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/helpconfig.py", line 104, in pytest_cmdline_parse
    @pytest.hookimpl(wrapper=True)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_result.py", line 103, in get_result
    raise exc.with_traceback(tb)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 38, in run_old_style_hookwrapper
    res = yield
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_callers.py", line 121, in _multicall
    res = hook_impl.function(*args)
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/config/__init__.py", line 1075, in pytest_cmdline_parse
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/config/__init__.py", line 1425, in parse
    finally:
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/config/__init__.py", line 1305, in _preparse
    for name in _iter_rewritable_modules(package_files):
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pluggy/_manager.py", line 416, in load_setuptools_entrypoints
    plugin = ep.load()
  File "/Users/coffeegist/.pyenv/versions/3.10.18/lib/python3.10/importlib/metadata/__init__.py", line 171, in load
    module = import_module(match.group('module'))
  File "/Users/coffeegist/.pyenv/versions/3.10.18/lib/python3.10/importlib/__init__.py", line 126, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1050, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1027, in _find_and_load
  File "<frozen importlib._bootstrap>", line 992, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "<frozen importlib._bootstrap>", line 1050, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1027, in _find_and_load
  File "<frozen importlib._bootstrap>", line 1006, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 688, in _load_unlocked
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/assertion/rewrite.py", line 186, in exec_module
    it.
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pytest_asyncio/__init__.py", line 7, in <module>
    from .plugin import fixture, is_async_test
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/_pytest/assertion/rewrite.py", line 186, in exec_module
    it.
  File "/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pytest_asyncio/plugin.py", line 38, in <module>
    from pytest import (
ImportError: cannot import name 'FixtureDef' from 'pytest' (/Users/coffeegist/Library/Caches/pypoetry/virtualenvs/bofhound-h-QmirpJ-py3.10/lib/python3.10/site-packages/pytest/__init__.py)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/lint/pylinter.py", line 752, in _lint_files
    self._lint_file(fileitem, module, check_astroid_module)
  File "/Users/coffeegist/.vscode/extensions/ms-python.pylint-2025.3.12271016/bundled/libs/pylint/lint/pylinter.py", line 790, in _lint_file
    raise astroid.AstroidError from e
astroid.exceptions.AstroidError

Expected behavior

No crash.

Pylint version

pylint 3.3.4
astroid 3.3.8
Python 3.10.18 (main, Sep 29 2025, 14:13:44) [Clang 17.0.0 (clang-1700.0.13.5)]

OS / Environment

darwin (Darwin)

[DanielNoord]

This should be raised against pylint_pytest as that is the package causing the import of pytest.

[Pierre-Sassoulas]

We can move this issue to pylint_pytest ourselves :) (but not me I'm on mobile right now)

[DanielNoord]

Moved it. Looking at the trace: potentially a pytest version mismatch or an import that should be guarded by a version check. FixtureDef might not always be exposed on the top level like that.