Plugin System

class BaseRequest:
    """
    Abstract base class for protocol-specific request implementations.
    """
    
    def __init__(
        self, 
        session: Any, 
        rspec: dict, 
        test_block_config: TestConfig
    ) -> None:
        """
        Initialize request with session, specification, and configuration.
        
        Parameters:
        - session: Protocol-specific session object
        - rspec: Request specification dictionary from YAML
        - test_block_config: Test configuration and variables
        """
    
    @property
    def request_vars(self) -> box.Box:
        """
        Variables used in the request for templating in subsequent stages.
        
        Returns:
        box.Box: Boxed request variables accessible via dot notation
        """
    
    def run(self):
        """
        Execute the request and return response.
        
        Returns:
        Protocol-specific response object
        """

@dataclasses.dataclass
class BaseResponse:
    """
    Abstract base class for protocol-specific response verification.
    """
    
    name: str
    expected: Any
    test_block_config: TestConfig
    response: Optional[Any] = None
    validate_functions: list[Any] = dataclasses.field(init=False, default_factory=list)
    errors: list[str] = dataclasses.field(init=False, default_factory=list)
    
    def verify(self, response):
        """
        Verify response against expected values and return saved variables.
        
        Parameters:
        - response: Actual response object from request execution
        
        Returns:
        Dictionary of variables to save for future test stages
        
        Raises:
        TestFailError: If response verification fails
        """
    
    def recurse_check_key_match(
        self,
        expected_block: Optional[Mapping],
        block: Mapping,
        blockname: str,
        strict: StrictOption,
    ) -> None:
        """
        Recursively validate response data against expected data.
        
        Parameters:
        - expected_block: Expected data structure
        - block: Actual response data
        - blockname: Name of the block being validated (for error messages)
        - strict: Strictness level for validation
        """

class TavernRestPlugin:
    """
    HTTP/REST protocol plugin using requests library.
    """
    
    session_type = "requests.Session"
    request_type = "RestRequest" 
    request_block_name = "request"
    verifier_type = "RestResponse"
    response_block_name = "response"

[project.entry-points.tavern_http]
requests = "tavern._plugins.rest.tavernhook:TavernRestPlugin"

test_name: HTTP API test
stages:
  - name: Make HTTP request
    request:  # Maps to request_block_name
      url: https://api.example.com/users
      method: POST
      json:
        name: "John Doe"
        email: "john@example.com"
      headers:
        Authorization: "Bearer {token}"
    response:  # Maps to response_block_name
      status_code: 201
      json:
        id: !anyint
        name: "John Doe"
      save:
        json:
          user_id: id

# MQTT Plugin Configuration
session_type = "MQTTClient"
request_type = "MQTTRequest"
request_block_name = "mqtt_publish"
verifier_type = "MQTTResponse" 
response_block_name = "mqtt_response"

[project.entry-points.tavern_mqtt]
paho-mqtt = "tavern._plugins.mqtt.tavernhook"

test_name: MQTT publish/subscribe test
stages:
  - name: Publish and verify message
    mqtt_publish:  # Maps to request_block_name
      topic: "sensor/temperature"
      payload:
        value: 23.5
        unit: "celsius"
        timestamp: "{timestamp}"
      qos: 1
    mqtt_response:  # Maps to response_block_name
      topic: "sensor/temperature/ack"
      payload:
        status: "received"
        message_id: !anystr
      timeout: 5

# gRPC Plugin Configuration
session_type = "GRPCClient"
request_type = "GRPCRequest"
request_block_name = "grpc_request"
verifier_type = "GRPCResponse"
response_block_name = "grpc_response"

[project.entry-points.tavern_grpc]
grpc = "tavern._plugins.grpc.tavernhook"

test_name: gRPC service test
stages:
  - name: Call gRPC method
    grpc_request:  # Maps to request_block_name
      service: "UserService"
      method: "CreateUser"
      body:
        name: "John Doe"
        email: "john@example.com"
      metadata:
        authorization: "Bearer {token}"
    grpc_response:  # Maps to response_block_name
      status: "OK"
      body:
        id: !anyint
        name: "John Doe"
        created_at: !anything

from tavern.request import BaseRequest
import box

class CustomProtocolRequest(BaseRequest):
    def __init__(self, session, rspec, test_block_config):
        self.session = session
        self.rspec = rspec  
        self.test_block_config = test_block_config
        self._request_vars = {}
    
    @property
    def request_vars(self) -> box.Box:
        return box.Box(self._request_vars)
    
    def run(self):
        # Implement protocol-specific request logic
        endpoint = self.rspec['endpoint']
        data = self.rspec.get('data', {})
        
        # Store variables for templating
        self._request_vars.update({
            'endpoint': endpoint,
            'request_time': time.time()
        })
        
        # Execute request and return response
        return self.session.send_request(endpoint, data)

from tavern.response import BaseResponse

@dataclasses.dataclass
class CustomProtocolResponse(BaseResponse):
    def verify(self, response):
        saved_variables = {}
        
        # Validate response
        if self.expected.get('success') is not None:
            if response.success != self.expected['success']:
                self._adderr("Expected success=%s, got %s", 
                           self.expected['success'], response.success)
        
        # Save variables for future stages
        if 'save' in self.expected:
            for key, path in self.expected['save'].items():
                saved_variables[key] = getattr(response, path)
        
        if self.errors:
            raise TestFailError(
                f"Response verification failed:\n{self._str_errors()}"
            )
        
        return saved_variables

from tavern._core.plugins import PluginHelperBase

class CustomProtocolPlugin(PluginHelperBase):
    session_type = CustomProtocolSession
    request_type = CustomProtocolRequest
    request_block_name = "custom_request"
    verifier_type = CustomProtocolResponse  
    response_block_name = "custom_response"

# setup.py or pyproject.toml
entry_points = {
    'tavern_custom': {
        'my_protocol = mypackage.plugin:CustomProtocolPlugin'
    }
}

# Use custom plugin via CLI
tavern-ci --tavern-custom-backend my_protocol test.yaml

# Or programmatically
from tavern.core import run
run("test.yaml", tavern_custom_backend="my_protocol")

# Tavern discovers plugins via entry points
import pkg_resources

def discover_plugins():
    plugins = {}
    for entry_point in pkg_resources.iter_entry_points('tavern_custom'):
        plugins[entry_point.name] = entry_point.load()
    return plugins

# HTTP plugin uses requests.Session
session = requests.Session()
session.headers.update({'User-Agent': 'Tavern/2.17.0'})

# MQTT plugin uses paho-mqtt client
client = mqtt.Client()
client.on_connect = handle_connect
client.connect(broker_host, broker_port)

# Custom plugin session
class CustomSession:
    def __init__(self, **config):
        self.connection = create_connection(config)
    
    def send_request(self, endpoint, data):
        return self.connection.call(endpoint, data)

# 1. Plugin creates session
session = plugin.session_type(**session_config)

# 2. Plugin creates request
request = plugin.request_type(session, request_spec, test_config)

# 3. Execute request
response = request.run()

# 4. Plugin creates response verifier  
verifier = plugin.verifier_type(
    name=stage_name,
    expected=expected_spec,
    test_block_config=test_config,
    response=response
)

# 5. Verify response and extract variables
saved_vars = verifier.verify(response)

class CustomProtocolError(TavernException):
    """Custom protocol-specific errors."""
    pass

class CustomRequest(BaseRequest):
    def run(self):
        try:
            return self.session.send_request(...)
        except ConnectionError as e:
            raise CustomProtocolError(f"Connection failed: {e}")
        except TimeoutError as e:
            raise CustomProtocolError(f"Request timeout: {e}")

@dataclasses.dataclass
class DatabaseRequest(BaseRequest):
    def run(self):
        query = self.rspec['query']
        params = self.rspec.get('params', {})
        
        cursor = self.session.execute(query, params)
        return cursor.fetchall()

@dataclasses.dataclass  
class DatabaseResponse(BaseResponse):
    def verify(self, rows):
        if 'row_count' in self.expected:
            if len(rows) != self.expected['row_count']:
                self._adderr("Expected %d rows, got %d", 
                           self.expected['row_count'], len(rows))
        
        return {'result_count': len(rows), 'first_row': rows[0] if rows else None}

# Usage in YAML:
# db_query:
#   query: "SELECT * FROM users WHERE age > ?"
#   params: [18] 
# db_response:
#   row_count: !anyint

import dataclasses
from typing import Any, Optional, Mapping
from abc import abstractmethod
import box

TestConfig = "tavern._core.pytest.config.TestConfig"
StrictOption = "tavern._core.strict_util.StrictOption"
PluginHelperBase = "tavern._core.plugins.PluginHelperBase"
TestFailError = "tavern._core.exceptions.TestFailError"