lager.factory

Module Contents

Classes

MyEncoder

Extensible JSON <http://json.org> encoder for Python data structures.

Step

Functions

hash_file(path)

get_secret(name[, default])

get_default_gateway_linux()

expand_choices(choices)

make_header(session_id, state)

Create header for bridge

start_data_session(session_id)

close_sock(sock, session_id)

readexactly(sock, num_bytes)

read_element(sock)

send_thread_function(sock, send_queue)

recv_thread_function(session_id, recv_queue, send_queue)

run(steps[, finalizer_cls])

Attributes

_STREAM_PORT

lager.factory._STREAM_PORT = 10120
exception lager.factory.Failure

Bases: Exception

Common base class for all non-exit exceptions.

lager.factory.hash_file(path)
lager.factory.get_secret(name, default=None)
lager.factory.get_default_gateway_linux()
class lager.factory.MyEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)

Bases: json.JSONEncoder

Extensible JSON <http://json.org> encoder for Python data structures.

Supports the following objects and types by default:

Python

JSON

dict

object

list, tuple

array

str

string

int, float

number

True

true

False

false

None

null

To extend this to recognize other objects, subclass and implement a .default() method with another method that returns a serializable object for o if possible, otherwise it should call the superclass implementation (to raise TypeError).

default(obj)

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
lager.factory.expand_choices(choices)
class lager.factory.Step(state, recv_queue=None, send_queue=None)
StopOnFail = True
output(event_type, event_data=None)
log(to_log, file=sys.stdout)
start()
done(result)
error(exc)
present_pass_fail_buttons(*, timeout=30)
present_buttons(buttons, *, timeout=30)
present_text_input(prompt, size=50, *, timeout=30)
update_heading(text)
present_select(label, choices, allow_multiple=False, *, timeout=30)
present_checkboxes(label, choices, *, timeout=30)
present_radios(label, choices, *, timeout=30)
update_image(filename)
hide_image()
_read_response(timeout=30)
lager.factory.make_header(session_id, state)

Create header for bridge

lager.factory.start_data_session(session_id)
lager.factory.close_sock(sock, session_id)
lager.factory.readexactly(sock, num_bytes)
lager.factory.read_element(sock)
lager.factory.send_thread_function(sock, send_queue)
lager.factory.recv_thread_function(session_id, recv_queue, send_queue)
lager.factory.run(steps, finalizer_cls=None)