pytest_test_categories.adapters.database¶
Production database blocker adapter using patching.
This module provides the production implementation of DatabaseBlockerPort that actually intercepts database connections by patching connect functions from database libraries.
The DatabasePatchingBlocker follows hexagonal architecture principles: - Implements the DatabaseBlockerPort interface (port) - Patches database connection functions to intercept access attempts - Raises DatabaseViolationError on unauthorized access - Restores original functions on deactivation
Patched Libraries: - sqlite3.connect (standard library, always available)
Optional libraries (patched only if installed): - psycopg2.connect / psycopg.connect (PostgreSQL) - pymysql.connect (MySQL) - pymongo.MongoClient (MongoDB) - redis.Redis / redis.StrictRedis (Redis) - sqlalchemy.create_engine (SQLAlchemy)
Example
>>> blocker = DatabasePatchingBlocker()
>>> try:
... blocker.activate(TestSize.SMALL, EnforcementMode.STRICT)
... # Any database connection will now be intercepted
... finally:
... blocker.deactivate() # Restore original database behavior
See also
DatabaseBlockerPort: The abstract interface in ports/database.py
FakeDatabaseBlocker: Test adapter in adapters/fake_database.py
SocketPatchingNetworkBlocker: Similar production adapter pattern for network
Classes¶
Production adapter that patches database connection functions to block access. |
Module Contents¶
- class pytest_test_categories.adapters.database.DatabasePatchingBlocker(/, **data)[source]¶
Bases:
pytest_test_categories.ports.database.DatabaseBlockerPortProduction adapter that patches database connection functions to block access.
This adapter intercepts database access by patching: - sqlite3.connect (standard library) - Optional: psycopg2.connect, psycopg.connect, pymysql.connect, etc.
The patching is reversible - deactivate() restores the original functions.
- Parameters:
data (Any)
Warning
This adapter modifies global state (database connection functions). Always use in a try/finally block or context manager to ensure cleanup.
Example
>>> blocker = DatabasePatchingBlocker() >>> try: ... blocker.activate(TestSize.SMALL, EnforcementMode.STRICT) ... sqlite3.connect(':memory:') # Raises DatabaseViolationError ... finally: ... blocker.deactivate()
Create a new model by parsing and validating input data from keyword arguments.
Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.
self is explicitly positional-only to allow self as a field name.
- model_post_init(context, /)[source]¶
Initialize post-Pydantic setup, storing references to original functions.
- Parameters:
context (object)
- Return type:
None
- reset()[source]¶
Reset blocker to initial state, restoring original database functions.
This is safe to call regardless of current state.
- Return type:
None
- current_enforcement_mode: pytest_test_categories.ports.network.EnforcementMode | None = None[source]¶
- current_test_size: pytest_test_categories.types.TestSize | None = None[source]¶