Skip to main content

luna.engine.engine

Engine Objects​

class Engine(EngineApi)

Local engine that mimics the real one for local testing purposes. It provides methods to add, remove, and tick contracts, handle events, manage secrets, and perform various get and set operations for templates, variables, parameters, roles, clauses, and more. Also includes functionalities to validate contracts and run functions, leveraging the local kernel and storage systems. This engine serves as a comprehensive testing tool for developing and validating contracts within the Luna framework.

add_contract​

def add_contract(*,
                 template: Template,
                 parameters: Dict[str, Any] | None = None,
                 linked_contracts: Dict[str, str] | None = None,
                 roles: Dict[str, List[str]] | None = None) -> ContractWrapper

Adds a contract to the engine using a given template and optional parameters. Returns a wrapper object for the contract.

add_contract_from_boundary​

def add_contract_from_boundary(
        *,
        template: str,
        parameters: Dict[str, JsonValue] | None = None,
        linked_contracts: Dict[str, str] | None = None,
        roles: Dict[str, List[str]] | None = None,
        class_name: str | None = None) -> ContractWrapper

Alternative method to add a contract using raw boundary data, suitable for API interactions.

rm_contract​

def rm_contract(contract_id: str)

Removes a specified contract from the engine.

tick_contract​

def tick_contract(contract_id: str, clock_offset: timedelta) -> TickData

Simulates a tick on a specified contract, advancing its state based on the given clock offset. Returns data generated during the tick.

put_event​

def put_event(contract_id: str,
              event_name: str,
              source_type: types.EventSourceType,
              source_id: str,
              publish_at: datetime | None = None,
              payload: PayloadDict | None = None)

Inserts an event into a contract, potentially triggering clauses.

put_audit​

def put_audit(
        req: engine_types.PutAuditRequest) -> engine_types.PutAuditResponse

Logs audit information for contract executions.

get_template​

def get_template(
        req: engine_types.GetTemplateRequest
) -> engine_types.GetTemplateResponse

Retrieves a contract template based on the provided request.

get_variable​

def get_variable(
        req: engine_types.GetVariableRequest
) -> engine_types.GetVariableResponse

Retrieves the value of a variable from a contract.

set_secret​

def set_secret(key: str, value: str)

Stores a secret value associated with a given key.

get_secret​

def get_secret(
        req: engine_types.GetSecretRequest) -> engine_types.GetSecretResponse

Retrieves a secret value based on a given key.

get_connection​

def get_connection(
    req: engine_types.GetConnectionRequest
) -> engine_types.GetConnectionResponse

Returns details of a connection.

This works by using the "user entered value" connection type to fake connections to return a specific value. When using this engine, each connection must have a value set using set_connection_read_value before that connection can be used.

set_connection_read_value​

def set_connection_read_value(connection_name: str, value: Any)

Stores the value to be returned by a connection.

set_participant_detail​

def set_participant_detail(participant_id: str, name: str, value: str)

Stores a secret value associated with a given key.

get_participant_detail​

def get_participant_detail(
    req: engine_types.GetParticipantDetailRequest
) -> engine_types.GetParticipantDetailResponse

Retrieves a secret value based on a given key.

update_participant_detail​

def update_participant_detail(
    req: engine_types.UpdateParticipantDetailRequest
) -> engine_types.UpdateParticipantDetailResponse

Retrieves a secret value based on a given key.

get_parameter​

def get_parameter(
    req: engine_types.GetParameterRequest
) -> engine_types.GetParameterResponse

Fetches the value of a specified parameter from a contract.

get_role​

def get_role(req: engine_types.GetRoleRequest) -> engine_types.GetRoleResponse

Retrieves role data, such as participant IDs, from a contract.

get_clause_eval_count​

def get_clause_eval_count(
    req: engine_types.GetClauseEvalCountRequest
) -> engine_types.GetClauseEvalCountResponse

Retrieves the evaluation count of a specified clause within a contract.

get_evaluated_clause​

def get_evaluated_clause(
    req: engine_types.GetEvaluatedClauseRequest
) -> engine_types.GetEvaluatedClauseResponse

Retrieves data for a clause that has been evaluated in a contract.

get_linked_contract​

def get_linked_contract(
    req: engine_types.GetLinkedContractRequest
) -> engine_types.GetLinkedContractResponse

Retrieves information about a contract linked to another contract.

get_children​

def get_children(
        req: engine_types.GetChildrenRequest
) -> engine_types.GetChildrenResponse

Obtains information about child contracts linked to a parent contract.

get_tick_count​

def get_tick_count(
    req: engine_types.GetTickCountRequest
) -> engine_types.GetTickCountResponse

Fetches the tick count, representing the number of iterations, of a contract.

get_tick_datetime​

def get_tick_datetime(
    req: engine_types.GetTickDatetimeRequest
) -> engine_types.GetTickDatetimeResponse

Retrieves the datetime of the most recent tick in a contract.

get_terminated​

def get_terminated(
    req: engine_types.GetTerminatedRequest
) -> engine_types.GetTerminatedResponse

Determines if a contract is terminated.

get_events​

def get_events(
        req: engine_types.GetEventsRequest) -> engine_types.GetEventsResponse

Fetches event data for a specific contract.

get_event​

def get_event(
        req: engine_types.GetEventRequest) -> engine_types.GetEventResponse

Retrieves a specific event based on its ID in a contract.

get_contract_info​

def get_contract_info(
    req: engine_types.GetContractInfoRequest
) -> engine_types.GetContractInfoResponse

Obtains comprehensive contract information, including state snapshot.

run_function​

def run_function(
        req: engine_types.RunFunctionRequest
) -> engine_types.RunFunctionResponse

Executes a specified function within the context of a contract.

validate​

def validate(
        req: engine_types.ValidateRequest) -> engine_types.ValidateResponse

Validates contract data based on the provided request. Ensures that parameters, accounts, and participants are correct.