/usr/lib/python3/dist-packages/plainbox/impl/transport.py is in python3-plainbox 0.5.3-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | # This file is part of Checkbox.
#
# Copyright 2013 Canonical Ltd.
# Written by:
# Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
# Daniel Manrique <roadmr@ubuntu.com>
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
"""
:mod:`plainbox.impl.transport` -- shared code for test data transports
======================================================================
.. warning::
THIS MODULE DOES NOT HAVE STABLE PUBLIC API
"""
from abc import ABCMeta, abstractmethod
from collections import OrderedDict
from logging import getLogger
import pkg_resources
from plainbox.abc import ISessionStateTransport
from plainbox.i18n import gettext as _
logger = getLogger("plainbox.transport")
class TransportError(Exception):
"""
Base class for problems encountered by the any ISessionStateTransport
during execution.
"""
class TransportBase(ISessionStateTransport):
"""
Base class for transports that send test data somewhere.
They handle just the transmission portion of data sending; exporters are
expected to produce data in the proper format (e.g. json, xml).
Each transport can have specific parameters that are required for the
other end to properly process received information (like system
identification, authorization data and so on), and that don't semantically
belong in the test data as produced by the exporter. Additionally
each transport needs to be told *where* to send test data. This is
transport-dependent; things like a HTTP endpoint, IP address, port
are good examples.
"""
def __init__(self, where, option_string):
self.url = where
#parse option string only if there's at least one k=v pair
self.options = {}
if not option_string:
return
if "=" in option_string:
self.options = {
option: value for (option, value) in
[pair.split("=", 1) for pair in option_string.split(",")]
}
if not self.options:
raise ValueError(_("No valid options in option string"))
def get_all_transports():
"""
Discover and load all transport classes.
Returns a map of transports (mapping from name to transport class)
"""
transport_map = OrderedDict()
iterator = pkg_resources.iter_entry_points('plainbox.transport')
for entry_point in sorted(iterator, key=lambda ep: ep.name):
try:
transport_cls = entry_point.load()
except ImportError as exc:
logger.exception(_("Unable to import {}: {}"), entry_point, exc)
else:
transport_map[entry_point.name] = transport_cls
return transport_map
|