Reference
Modules
Parse bombard command line args.
Attributes
module-attribute
CAMPAIGN_FILE_NAME = 'bombard.yaml'
module-attribute
DIR_DESC_FILE_NAME = 'README.md'
module-attribute
EXAMPLES_PREFIX = 'bombard://'
module-attribute
INIT_EXAMPLE = 'easy.yaml'
module-attribute
REPEAT = 10
module-attribute
THREADS_NUM = 10
module-attribute
THRESHOLD = 1000
module-attribute
TIMEOUT = 15
Functions
get_args() -> Any
Dict with access to keys as attributes.
d = AttrDict({'a': 'b', 'c': 'g'}) d.c 'g'
Instance can be used as callable to set value to the dict
d = AttrDict({}) d(k='t') d.k 't'
(!) All values that you set will be repeated to master that you used to init the object.
master = {'j': 'l'} d = AttrDict(master) d(h='q') master['h'] 'q' d['w'] = 'e' master['w'] 'e'
And it works without master
d = AttrDict() d(f=7) d.f 7
Classes
Bases: dict[str, Any]
You can access all dict values as attributes. All changes immediately repeated in master_dict.
instance-attribute
master_dict = master_dict
Attributes
module-attribute
AMMO = 'ammo'
module-attribute
DEFAULT_OK = {200}
module-attribute
DEFAULT_OVERLOAD = [502, 504]
module-attribute
PREPARE = 'prepare'
module-attribute
log = getLogger()
Classes
Bases: WeaverMill
Use horde of threads to make HTTP-requests
instance-attribute
args = args
instance-attribute
campaign = campaign_book
instance-attribute
ok = ok_statuses if ok_statuses is not None else DEFAULT_OK
instance-attribute
overload = (
overload_statuses
if overload_statuses is not None
else DEFAULT_OVERLOAD
)
instance-attribute
reporter = Reporter(
time_units="ms" if ms else None,
time_threshold_ms=int(threshold),
success_statuses=ok,
)
instance-attribute
request_fired = False
instance-attribute
resp_count = 0
instance-attribute
show_request = {1: 'Sent 1st request..'}
instance-attribute
show_response = {1: 'Got 1st response..'}
instance-attribute
supply = supply if supply is not None else {}
staticmethod
beautify_url(
url: str, method: str, body: Optional[str]
) -> str
staticmethod
get_headers(
request: dict[str, Any], body_is_json: bool
) -> dict[str, Any]
Treat special value 'json' as Content-Type: application/json
process_resp(
ammo: dict[str, Any],
status: Union[int, str],
resp: str,
elapsed: int,
size: int,
) -> None
reload(
requests: Any,
repeat: Optional[int] = None,
prepare: bool = False,
**kwargs: Any
) -> None
Add request(s) to the bombardier queue repeat-times (args.repeat if None).
If repeat field exists in the request additionally repeats as defined by it.
Requests can be one request or list of requests. If supply specified it'll be used in addition to self.supply.
Arg prepare indicate call from main, not from request script.
So we know if any scripts call reload (self.request_fired)
report() -> None
status_coloured(status: Union[str, int]) -> str
Paint ok status as green and overload as red using terminal control codes.
If status has special string value (EXCEPTION_STATUS) paint it red.
worker(thread_id: int, job: dict[str, Any]) -> None
Thread callable. Strike ammo from queue.
Functions
apply_supply(s: str, supply: dict[str, Any]) -> str
apply_supply_dict(
request: dict[str, Any], supply: dict[str, Any]
) -> dict[str, Any]
Use supply to substitute all {name} in request strings.
Modules
Bombard campaign loader.
Extends yaml loader with loading external files !include file.ext.
Excludes lines that import mock_globals.
Attributes
module-attribute
SIGNATURE = 'bombard.mock_globals'
Classes
Bases: SafeLoader
include(node)
staticmethod
wrap_in_yaml_document(msg: str) -> str
Converts multi-line msg to yaml document that we can insert into yaml
staticmethod
full_load(stream: Any, Loader: Any = None) -> Any
Mimics yaml interface for seamless injection
staticmethod
load(stream: Any, Loader: Any = None) -> Any
Mimics yaml interface for seamless injection
Attributes
Functions
expand_relative_file_name(file_name: str) -> str
Replace RELATIVE_PREFIX with package folder.
so bombard script can use internal examples without full path spec
get_campaign_file_name(args: Any) -> str
show_folder(folder_path: str) -> None
Attributes
module-attribute
EXCEPTION_STATUS = '!!!'
Functions
http_request(
url: str,
method: str = "GET",
headers: Optional[dict[str, Any]] = None,
body: Optional[str] = None,
timeout: Optional[int] = None,
) -> tuple[Union[int, str], Any]
Make HTTP request.
Returns tuple:
Bombard's main
Attributes
Classes
Functions
add_names_to_requests(
campaign_book: dict[str, Any],
) -> None
Duplicate names inside requests so worker will see it and use in stat report
campaign(args: Any) -> None
get_supply_from_cli(
supply: Optional[list[str]],
) -> dict[str, Any]
Extract key=value pairs from list of supply args
guess_type(value: str) -> Union[str, int, float]
Converts value in int or float if possible
init(args: Any) -> None
Copies the example to current folder as bombard.yaml
load_book_supply(
cli_supply: dict[str, Any], book_supply: dict[str, Any]
) -> None
Updates CLI supply with supply from campaign book. But do not overwrite CLI supply with book supply - values from CLI have bigger priority.
main() -> None
start_campaign(
args: Any, campaign_book: dict[str, Any]
) -> None
If you use include files for scripts add into them
from bombard.mock_globals import *; master(<your yaml>)
That defines globals so you have valid code and code autocomplete in your IDE editor.
All strings with bombard.examples.mock_globals will be automatically removed before running bombard scripts.
Attributes
module-attribute
ammo = _Ammo()
module-attribute
args = _Args()
module-attribute
resp = {}
module-attribute
supply = _Supply()
Functions
master(yaml_file_name: str)
Add names from the yaml file to the unit globals. That makes code autocomplete work in bombard script.
reload(requests, repeat=None, **kwargs)
- time_ns for Python before 3.7
- Elapsed context manager
- pretty_ns to represent time elapsed in human lovable form
Usage:
start = time_ns() import time time.sleep(0.0000001) pretty_ns((start + 100) - start) '0.1 mks'
with Timer() as timer: ... import time ... time.sleep(0.0000001) ... timer.pretty.endswith('s') ... timer.ns > 0 True True
Attributes
module-attribute
time_ns = time_ns
Classes
property
ns: int
property
pretty: str
instance-attribute
start: int
Functions
emul_time_ns() -> int
pretty_ns(
elapsed_ns: int, fixed_units: Optional[str] = None
) -> str
for earlier Python versions this is emulation of the Python3.7 time_ns
Functions
pretty_sz(size: Union[int, float]) -> str
Bombard reporter.
Use:
* log to add each request result.
* report to generate report.
Attributes
module-attribute
ARRAY_UINT32 = 'L'
module-attribute
ARRAY_UINT64 = 'Q'
module-attribute
FAIL_GROUP = 'fail'
module-attribute
SIZE = 'size'
module-attribute
SUCCESS_GROUP = 'success'
module-attribute
TIME = 'time'
module-attribute
TIME_DENOMINATOR = 1
Classes
Report bombard's result
Access to stat data only by log(), filter(), reduce()
instance-attribute
DIMENSIONS: dict[str, dict[str, Any]] = {
TIME: {
"type": ARRAY_UINT64,
"pretty_func": pretty_time,
},
SIZE: {"type": ARRAY_UINT32, "pretty_func": pretty_sz},
}
instance-attribute
STAT_DEFAULT: dict[str, MutableSequence[Any]] = {
name: array(params["type"])
for (name, params) in items()
}
instance-attribute
ok = success_statuses or {200}
instance-attribute
start_ns = time_ns()
instance-attribute
stat: dict[
str, dict[int, dict[str, MutableSequence[Any]]]
] = {}
instance-attribute
time_threshold_ns = time_threshold_ms * 10 ** 6
instance-attribute
time_units = time_units
property
total_elapsed_ns: int
staticmethod
dimension_stat_report(
dimension_values: MutableSequence[Union[int, float]],
pretty_func: Callable[[Union[int, float]], str],
) -> str
filter(
dimension_name: str,
status_group_filter: Optional[str] = None,
request_name_filter: Optional[str] = None,
) -> MutableSequence[Any]
Filter the dimension by group and/or request_name, Returns array with values from dimention_name
filtered_report(
status_group_filter: Optional[str] = None,
request_name_filter: Optional[str] = None,
) -> str
Filter by group and/or request_name. Returns report str with stats for all dimensions.
group_name_by_status(status: int) -> str
All this statuses should be in GROUPS
log(
status: Union[int, str],
elapsed: int,
request_name: str,
response_size: int,
) -> None
Add result to the report
:param status: HTTP response status :param elapsed: Request-Response time, ns :param request_name: Request name or None :param response_size: Response body size
pretty_ns(elapsed_ns: int, paint: bool = True) -> str
pretty_time(elapsed: int, paint: bool = True) -> str
reduce(
reduce_func: Callable[[Any], Any],
dimension_name: str,
status_group_filter: Optional[str] = None,
request_name_filter: Optional[str] = None,
) -> int
Reduce the dimension by group and/or request_name with the reduce_func Returns dict {'time':, 'size':}
report() -> str
statuses_report(
request_name_filter: Optional[str] = None,
) -> str
Functions
Modules
Logger with thread storage to add to log information about request.
Use main_thread/sending/receiving to switch logger modes.
Attributes
module-attribute
log = getLogger()
module-attribute
thread_data = local()
Classes
Bases: Formatter
format(record: Any) -> str
Functions
main_thread() -> None
We are in main thread
receiving() -> None
Got response to request
sending(
thread_id: int, request_id: str, request_name: str
) -> None
Start sending request
setup_logging(
level: int, log_file_name: Optional[str] = None
) -> None
Functions
markdown_for_terminal(descr: str) -> str
Colourize text in terminal source https://en.wikipedia.org/wiki/ANSI_escape_code#Colors
You can use it function style
green('Hello!') '\x1b[1;32mHello!\x1b[0m'
Or include style
f'{YELLOW}Hello{OFF}, {RED}world{OFF}!' '\x1b[1;33mHello\x1b[0m, \x1b[1;31mworld\x1b[0m!'
Under the hood this is colorama. But I keep my wrapper in this module as legacy.
Attributes
module-attribute
BROWN = code_to_chars(f'{DIM};{YELLOW}')
module-attribute
DARK_RED = code_to_chars(f'{DIM};{RED}')
module-attribute
GRAY = code_to_chars(f'{BRIGHT};{BLACK}')
module-attribute
GREEN = code_to_chars(f'{BRIGHT};{GREEN}')
module-attribute
OFF = code_to_chars(RESET_ALL)
module-attribute
RED = code_to_chars(f'{BRIGHT};{RED}')
module-attribute
YELLOW = code_to_chars(f'{BRIGHT};{YELLOW}')
Functions
brown(s: str) -> str
dark_red(s: str) -> str
gray(s: str) -> str
green(s: str) -> str
paint_it(msg: str, colour: str) -> str
red(s: str) -> str
yellow(s: str) -> str
Attributes
module-attribute
VERSION = '1.20.5'
Multithreading jobs processor abstraction.
Override method worker in descendant to do a job.
Add jobs with put.
Start processing with start.
Classes
instance-attribute
job_count = 0
instance-attribute
queue: Queue[Optional[dict[str, Any]]] = Queue()
instance-attribute
threads = []
instance-attribute
threads_num = threads_num
process() -> None
Starts all threads and lock until queue is empty
put(job: dict[str, Any]) -> None
Add job to queue.
To start processing use process.
stop() -> None
Stops all threads - send stop signal to queue and lock until they stop
thread_worker(thread_id: int) -> None
Get job from queue and pass it to abstract worker that should be implemented in descendant.
Even if worker throw exception we mark the job we gave him as done.
Job == None is a signal to stop work.
abstractmethod
worker(thread_id: int, job: dict[str, Any]) -> None
Implement your job processor, runs in thread.
:param thread_id: just sequential number of the thread we work into :param job: job from queue