Add StateMachinePlugin infrastructure and stateful protocol plugins (v0.4.0)#2
Merged
Add StateMachinePlugin infrastructure and stateful protocol plugins (v0.4.0)#2
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a significant enhancement to the Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a significant and powerful new feature with the StateMachinePlugin and a suite of stateful protocol plugins. The implementation is extensive and well-structured, and the documentation is thorough. I've found a couple of areas for improvement: one is a refactoring opportunity to reduce boilerplate in the new plugin proxies, and another is a potential correctness issue in the PopenPlugin's wait() implementation. There's also a small improvement for a code example in the new documentation. Overall, this is a fantastic addition to the library.
Note: Security Review did not run due to the size of the PR.
Comment on lines
+163
to
+170
| def wait(self, timeout: float | None = None) -> int: | ||
| plugin = _find_popen_plugin() | ||
| handle = plugin._lookup_session(self) | ||
| result = plugin._execute_step(handle, "wait", (), {}, _SOURCE_WAIT) | ||
| # result is returncode int | ||
| self.returncode = int(result) | ||
| plugin._release_session(self) | ||
| return self.returncode |
There was a problem hiding this comment.
The current implementation of wait() releases the session via plugin._release_session(self). This is problematic because the real subprocess.Popen.wait() can be called multiple times on a terminated process, and it should return the return code without error. Releasing the session here will cause subsequent calls to wait() or other methods on the same _FakePopen instance to fail with a _lookup_session error.
I suggest making the wait() method state-aware to handle multiple calls correctly and not release the session. The session should live as long as the Popen object itself.
Suggested change
| def wait(self, timeout: float | None = None) -> int: | |
| plugin = _find_popen_plugin() | |
| handle = plugin._lookup_session(self) | |
| result = plugin._execute_step(handle, "wait", (), {}, _SOURCE_WAIT) | |
| # result is returncode int | |
| self.returncode = int(result) | |
| plugin._release_session(self) | |
| return self.returncode | |
| def wait(self, timeout: float | None = None) -> int: | |
| plugin = _find_popen_plugin() | |
| handle = plugin._lookup_session(self) | |
| if handle._state == "terminated": | |
| if self.returncode is None: | |
| # This should not be reachable if state transitions are correct. | |
| raise RuntimeError("BUG: process terminated but returncode is None") | |
| return self.returncode | |
| result = plugin._execute_step(handle, "wait", (), {}, _SOURCE_WAIT) | |
| # result is returncode int | |
| self.returncode = int(result) | |
| # Do not release the session, to allow subsequent calls to wait() or poll(). | |
| return self.returncode |
| sock = socket.socket() | ||
| # Bind the session without connecting first by directly using _bind_connection: | ||
| from bigfoot.plugins.socket_plugin import SocketPlugin | ||
| plugin = bigfoot.current_verifier()._plugins[-1] |
There was a problem hiding this comment.
The example code for InvalidStateError uses plugin = bigfoot.current_verifier()._plugins[-1] to get the plugin instance. This is brittle as it assumes the SocketPlugin is the last one registered on the verifier. This could lead to flaky tests for users who copy this example if they have other plugins registered.
A more robust way to get the plugin would be to find it by its type.
Suggested change
| plugin = bigfoot.current_verifier()._plugins[-1] | |
| plugin = next(p for p in bigfoot.current_verifier()._plugins if isinstance(p, SocketPlugin)) |
Comment on lines
+191
to
+415
| class _SubprocessProxy: | ||
| """Proxy to the SubprocessPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _SubprocessPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _SubprocessPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _SubprocessPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| subprocess_mock = _SubprocessProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Popen proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _PopenProxy: | ||
| """Proxy to the PopenPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _PopenPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _PopenPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _PopenPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| popen_mock = _PopenProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # SMTP proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _SmtpProxy: | ||
| """Proxy to the SmtpPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _SmtpPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _SmtpPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _SmtpPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| smtp_mock = _SmtpProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Socket proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _SocketProxy: | ||
| """Proxy to the SocketPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _SocketPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _SocketPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _SocketPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| socket_mock = _SocketProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Database proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _DatabaseProxy: | ||
| """Proxy to the DatabasePlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _DatabasePlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _DatabasePlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _DatabasePlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| db_mock = _DatabaseProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # AsyncWebSocket proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _AsyncWebSocketProxy: | ||
| """Proxy to the AsyncWebSocketPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. Raises ImportError if the | ||
| websockets extra is not installed. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| from bigfoot.plugins.websocket_plugin import _WEBSOCKETS_AVAILABLE | ||
|
|
||
| if not _WEBSOCKETS_AVAILABLE: | ||
| raise ImportError( | ||
| "bigfoot[websockets] is required to use bigfoot.async_websocket_mock. " | ||
| "Install it with: pip install bigfoot[websockets]" | ||
| ) | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _AsyncWebSocketPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _AsyncWebSocketPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _AsyncWebSocketPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| async_websocket_mock = _AsyncWebSocketProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # SyncWebSocket proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _SyncWebSocketProxy: | ||
| """Proxy to the SyncWebSocketPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. Raises ImportError if the | ||
| websocket-client extra is not installed. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| from bigfoot.plugins.websocket_plugin import _WEBSOCKET_CLIENT_AVAILABLE | ||
|
|
||
| if not _WEBSOCKET_CLIENT_AVAILABLE: | ||
| raise ImportError( | ||
| "bigfoot[websocket-client] is required to use bigfoot.sync_websocket_mock. " | ||
| "Install it with: pip install bigfoot[websocket-client]" | ||
| ) | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _SyncWebSocketPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _SyncWebSocketPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _SyncWebSocketPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| sync_websocket_mock = _SyncWebSocketProxy() | ||
|
|
||
|
|
||
| # --------------------------------------------------------------------------- | ||
| # Redis proxy singleton | ||
| # --------------------------------------------------------------------------- | ||
|
|
||
|
|
||
| class _RedisProxy: | ||
| """Proxy to the RedisPlugin registered on the current test verifier. | ||
|
|
||
| Auto-creates the plugin on first access per test. Raises ImportError if | ||
| the redis extra is not installed. | ||
| """ | ||
|
|
||
| def __getattr__(self, name: str) -> object: | ||
| from bigfoot.plugins.redis_plugin import _REDIS_AVAILABLE | ||
|
|
||
| if not _REDIS_AVAILABLE: | ||
| raise ImportError( | ||
| "bigfoot[redis] is required to use bigfoot.redis_mock. " | ||
| "Install it with: pip install bigfoot[redis]" | ||
| ) | ||
| verifier = _get_test_verifier_or_raise() | ||
| plugin: _RedisPlugin | None = None | ||
| for p in verifier._plugins: | ||
| if isinstance(p, _RedisPlugin): | ||
| plugin = p | ||
| break | ||
| if plugin is None: | ||
| plugin = _RedisPlugin(verifier) | ||
| return getattr(plugin, name) | ||
|
|
||
|
|
||
| redis_mock = _RedisProxy() |
There was a problem hiding this comment.
There's a lot of duplicated code across the new proxy classes (_SubprocessProxy, _PopenProxy, etc.). The __getattr__ method in each is nearly identical. This could be refactored to reduce boilerplate and improve maintainability, for example by using a generic factory function or a base class for these proxies.
For example, a generic helper could look like this:
from typing import Type, TypeVar
P = TypeVar("P", bound="BasePlugin")
def _get_or_create_plugin(verifier: "StrictVerifier", plugin_type: Type[P]) -> P:
for p in verifier._plugins:
if isinstance(p, plugin_type):
return p
return plugin_type(verifier)This would simplify each proxy's __getattr__ implementation significantly.
Prevents git worktree directories from being tracked by the repository.
Adds InvalidStateError to _errors.py with source_id, method, current_state, and valid_states attributes, and exports it via __init__.py. Tests cover message format (exact equality), all four attributes, and BigfootError inheritance.
Add ScriptStep dataclass, SessionHandle, and StateMachinePlugin ABC to support state-machine-driven connection mocking. StateMachinePlugin manages session queues, connection binding, state transitions, and auto-marks interactions asserted to avoid UnassertedInteractionsError. Also fix test_init.py expected_all set to include InvalidStateError (added by Task 1.1 but missing from the test assertion). 54 new tests in tests/unit/test_state_machine_plugin.py. All 413 tests pass. ruff check and mypy --strict pass.
Adds SocketPlugin, a StateMachinePlugin subclass that intercepts socket.socket.connect/send/sendall/recv/close via reference-counted class-level monkey-patching. States: disconnected -> connected -> closed. - src/bigfoot/plugins/socket_plugin.py: new SocketPlugin - src/bigfoot/__init__.py: add socket_mock proxy and SocketPlugin to __all__ - tests/unit/test_socket_plugin.py: 16 new tests (all watched fail first) - tests/unit/test_init.py: update __all__ assertion for new public symbols
…tion Implements Task 3.1: DatabasePlugin intercepts sqlite3.connect(), returns _FakeConnection backed by _FakeCursor/_FakeCursorProxy. Supports execute, cursor, commit, rollback, close with full state-machine validation (connected -> in_transaction -> connected/closed). Adds bigfoot.db_mock proxy and DatabasePlugin to the public API. 23 new tests written TDD-first.
… (Tasks 4.1, 4.2) Add AsyncWebSocketPlugin (websockets library) and SyncWebSocketPlugin (websocket-client library) as state-machine-driven bigfoot plugins with full TDD. Both intercept at the module function level (websockets.connect and websocket.create_connection), support FIFO session ordering, reference- counted activation, and raise ImportError with installation hints when the optional dependency is absent. Add websockets and websocket-client as optional extras in pyproject.toml and include them in the dev sync group. Update bigfoot.__init__ with both plugins and their proxy singletons. Apply ruff format to pre-existing files that were not formatted under the current ruff version. 29 new tests in tests/unit/test_websocket_plugin.py. Full suite: 481 passed.
Implements Task 5.1. PopenPlugin intercepts subprocess.Popen via class replacement (_FakePopen), routing init/stdin.write/stdout.read/stderr.read/ communicate/wait through the StateMachinePlugin state machine. Includes reference-counted activation, conflict detection, and coexistence with SubprocessPlugin. Adds bigfoot.popen_mock proxy and PopenPlugin to __all__.
Add SmtpPlugin with class-replacement interception of smtplib.SMTP. Supports connect/ehlo/helo/starttls/login/sendmail/send_message/quit with reference-counted activate/deactivate. Add smtp_mock proxy to public API. 19 new tests cover all flows, error paths, and edge cases.
RedisPlugin extends BasePlugin directly (stateless command interception). Patches redis.Redis.execute_command at the class level with reference counting. Each command name (uppercase) has its own FIFO deque of RedisMockConfig objects. Adds redis_mock proxy and redis extra to pyproject.toml. 27 new tests; full suite: 551 passed.
…to 0.4.0
- _AsyncWebSocketProxy.__getattr__ now checks _WEBSOCKETS_AVAILABLE early and
raises ImportError("bigfoot[websockets] is required...") before attempting to
create a plugin, matching the pattern already used by _RedisProxy.
- _SyncWebSocketProxy.__getattr__ does the same for _WEBSOCKET_CLIENT_AVAILABLE.
- Two new TDD-first tests in test_init.py verify these error paths using monkeypatch.
- Version bumped from 0.3.0 to 0.4.0 (new public API surface added across Tracks 2-8).
Add docs/guides/stateful-plugins.md with a comprehensive tutorial covering SocketPlugin, DatabasePlugin, AsyncWebSocketPlugin, SyncWebSocketPlugin, PopenPlugin, SmtpPlugin, and RedisPlugin. Each section includes quickstart examples derived from the test suite, state machine diagrams, and common error explanations. Append a StateMachinePlugin section to writing-plugins.md covering when to use the state machine base class, all abstract methods, the session scripting API, and a minimal FtpPlugin implementation skeleton. Add [0.4.0] changelog entry listing all new plugins, StateMachinePlugin, InvalidStateError, and the three optional extras.
Remove the redundant _SOURCE_CURSOR_EXECUTE constant ("db:cursor:execute")
from DatabasePlugin and make _FakeCursorProxy.execute() use the same
_SOURCE_EXECUTE ("db:execute") as _FakeConnection.execute(). Both paths
represent the same logical operation and should produce consistent
source_ids in error messages and timeline records.
Add StateMachinePlugin._register_connection() to encapsulate the
_active_sessions/_connection_refs write that was previously duplicated
inline across both websocket plugins. Refactor _bind_connection() to
delegate to it, and update AsyncWebSocketPlugin and SyncWebSocketPlugin
to call _register_connection() instead of manipulating the private dicts
directly, restoring the abstraction boundary.
…pen, and smtp tests Replace tautological isinstance checks, banned substring-in checks, and len > 0 guards with exact equality assertions that actually verify correctness. Add missing source_id assertion on InvalidStateError and end-to-end result verification on redis_mock proxy test.
- Fix InvalidStateError example in stateful-plugins.md to match actual format string from _errors.py - Correct abstract method counts in writing-plugins.md: BasePlugin has ten abstract methods (not nine); StateMachinePlugin inherits seven from BasePlugin (not four) - Remove task number references from websocket_plugin.py module docstring and section comments - Strip historical TDD narrative from five test module docstrings - Expand _state_machine_plugin.py lifecycle docstring to note the WebSocket pattern of manual queue pop plus _register_connection - Add starttls self-loop note to SmtpPlugin class docstring
…kup, deduplicated proxy boilerplate - PopenPlugin.wait() is now idempotent: a second call returns self.returncode directly without consuming another script step or requiring an active session. The _release_session() call is removed from wait() since session lifetime is managed by the sandbox context, not by individual method calls. - Replace _plugins[-1] index lookup in docs/guides/stateful-plugins.md with a type-safe isinstance-based search using next(), matching the production pattern. - Extract _get_or_create_plugin(verifier, plugin_type) helper in __init__.py and refactor all nine proxy __getattr__ implementations to use it, eliminating the repeated for-loop-and-break boilerplate. TypeVar _T ensures the return type is correctly inferred by mypy under --strict.
elijahr
force-pushed
the
elijahr/stateful-plugins
branch
from
96d04b2 to
f0c0050
Compare
2 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
StateMachinePlugin— abstract base class for stateful protocol plugins providing a FIFO session queue, state-transition validation via_transitions(), and anew_session()/_bind_connection()/_execute_step()/_release_session()lifecycle with automatic interaction assertion at step execution timeInvalidStateError— raised when a method is called from a state not listed as a valid from-state; carriessource_id,method,current_state, andvalid_statesattributesStateMachinePlugin:SocketPlugin— mockssocket.socketwithdisconnected → connected → closedstate machine; accessible viabigfoot.socket_mockDatabasePlugin— mockssqlite3.connect()with full execute/commit/rollback/close lifecycle and fake cursor objects supportingfetchone(),fetchall(),fetchmany(), and iteration; accessible viabigfoot.db_mockAsyncWebSocketPlugin— mockswebsockets.connectasync context manager; requiresbigfoot[websockets]; accessible viabigfoot.async_websocket_mockSyncWebSocketPlugin— mockswebsocket.create_connection(websocket-client library); requiresbigfoot[websocket-client]; accessible viabigfoot.sync_websocket_mockPopenPlugin— mockssubprocess.Popenwith stdin/stdout/stderr stream scripting andcommunicate()/wait()lifecycle; coexists withSubprocessPlugin; accessible viabigfoot.popen_mockSmtpPlugin— mockssmtplib.SMTPwith optionalstarttls/loginbranches; supports authenticated and unauthenticated send flows; accessible viabigfoot.smtp_mockRedisPlugin— mocksredis.Redis.execute_commandwith per-command FIFO queues; stateless; requiresbigfoot[redis]; accessible viabigfoot.redis_mockbigfoot[websockets],bigfoot[websocket-client],bigfoot[redis]0.4.0Test plan
elijahr/stateful-plugins(uv run pytest -q)StateMachinePluginlifecycle and transition validation logicpyproject.tomland degrade gracefully when absent🤖 Generated with Claude Code