API
Module structure
A module must have a top level function named run
, that receives a single argument. athena will execute this function with an Athena
instance as the argument.
from athena.client import Athena
def run(athena: Athena):
client = athena.client()
client.get('http://echo.jsontest.com/key/value')
The function may also be async
, if an asynchronous context is desired.
from athena.client import Athena
async def run(athena: Athena):
client = athena.client()
tasks = [client.get_async("https://google.com") for _ in range(10)]
await asyncio.gather(*tasks)
Sending requests
Create a client
The injected Athena
instance provides methods to create and send requests. Start by creating a new Client
.
The client can be configured by providing a builder function. The builder will be applied to each request sent by the client. The configuration is done through the RequestBuilder
interface.
def run(athena: Athena):
client = athena.client(lambda requestBuilder: requestBuilder
.base_url("https://www.example.com")
.auth.bearer(my_token)
)
Send the request
The client provides methods to send requests, these methods can optionally include a configuration function. The configuration will be applied on top of the client configuration.
def run(athena: Athena):
client = athena.client(lambda requestBuilder: requestBuilder
.base_url("https://www.example.com")
.auth.bearer(my_token)
)
client.get('/foo')
client.post('/bar', lambda r: r
.body.json({
'baz': 'qux'
})
)
Parse the response
The client methods will return a ResponseTrace
, which contains information about the response.
def run(athena: Athena):
...
response = client.get('/foo')
print(f'status: {response.status_code} {response.reason}')
athena can provide more information about the rest of the request with the trace
method, which will return the AthenaTrace
for the whole request/response saga.
def run(athena: Athena):
...
trace = athena.trace()
print(f"request payload: {trace.request.text}")
print(f"request time: {trace.elapsed}")
Configuring the request
Hooks
athena can run pre-request and post-request hooks at the client or request level, using the HookStepFactory
.
def run(athena: Athena):
client = athena.client(lambda b: b
.hook.before(lambda r: print("I am about to send a request with these headers: ", r.headers))
.hook.after(lambda r: print("I just received a response with the reason:", r.reason))))
Other utilities
Environments, Variables and Secrets
athena will provide variables and secrets to the module under test through the Athena
object.
def run(athena: Athena):
# will return `None` if no such variable exists
username = athena.variable.get('username')
# will throw error if no such secret exists
password = athena.secret['password']
# will use default value if no such variable exists
email = athena.variable.get('email', default='foo@bar.baz')
# will force the type of the value to be a bool
is_admin = athena.variable.bool['is_admin']
This will reference the variables.yml
and secrets.yml
environment files. athena will select all variable or secret files that can be found in any ancestor directory of the module being run (up to the athena root directory). For example, if we are running the following module:
then athena will look for variables in the following locations:
Resource files
The format of both the secrets and variables files is the same, objects are defined as name.environment: value
.
Environments
By default, athena will use the __default__
environment, but you can specify one in the run
command.
You can also set the default environment.
Fixtures
athena supports adding fixtures using the same heirarchy strategy as the variables and secrets files. Any file named fixture.py
in a directory that is a direct ancestor of the current module will be loaded. The fixtures are executed in top-down order, meaning you can access fixtures created in an ancestor directory from fixtures inside a descendant one.
athena will call the fixture method on Athena.fixture
before running any modules. The fixture attribute is an anonymous object to which fixture methods can assign values or functions to. These values/functions are then available to the test module.
from athena.client import Fixture, Athena
def fixture(fixture: Fixture):
def build_client(athena: Athena):
base_url = athena.variable["base_url"]
api_key = athena.secret["api_key"]
client = athena.client(lambda b: b
.base_url(base_url)
.auth.bearer(api_key))
return client
fixture.some_value = 'some_value'
fixture.client = build_client
from athena.client import Athena
def run(athena: Athena):
some_value = athena.fixture.some_value
client = athena.fixture.client(athena)
client.post("path/to/resource")
Infix
athena also provides the infix
attribute, short for "into fixture".
This property is used similarly to fixture
, but it can only be called with fixtures that are functions. infix
will inject the Athena
instance into the fixture function as the first argument, and pass along the rest, making for
a useful shorthand.
from athena.client import Athena
def run(athena: Athena):
client = athena.infix.client()
client.post("path/to/resource")
Fake
The fake
attribute is a thin wrapper / extension around Faker. This allows you to generate randomized data for requests.
from athena.client import Athena
def run(athena: Athena):
client = athena.infix.client()
client.post("api/users", lambda r: r
.body.json({
'name': athena.fake.first_name()
})
)
Caching
athena provides a basic key (str
) - value (str
, int
, float
, bool
) cache. The cache is global and is persisted between runs.
import time
from athena.client import Athena
def refresh_token(athena: Athena):
if "token" not in athena.cache \
or "token_exp" not in athena.cache \
or athena.cache["token_exp"] < time.time():
athena.cache["token"], athena.cache["token_exp"] = athena.infix.get_token()
return athena.cache["token"]
def run(athena: Athena):
token = refresh_token(athena)
client = athena.infix.client(token)
client.get("path/to/resource")
the cache is persisted in the .cache
file, and can be cleared by deleting the file or from the cli:
Jsonification
athena provides a jsonify
tool to json-dump athena objects, like AthenaTrace
.
Apart from adding an encoder for athena objects, this method will pass-through arguments
like indent
to json.dumps
.
from athena.client import Athena, jsonify
def run(athena: Athena):
athena.client().get("http://haondt.com")
traces = athena.traces()
print(jsonify(traces, indent=4))
Context
the context
(Context
) property provides information about the runtime environment of the module.
from athena.client import Athena
def run(athena: Athena):
print("current module:", athena.context.module_name)
print("current environment:", athena.context.environment)
Assertions
athena comes bundled with a thin wrapper around the assert
statement called athert
. This wrapper provides
more informative error messages and a fluent syntax.
from athena.client import Athena, Client
from athena.test import athert
def run(athena: Athena):
client: Client = athena.infix.build_client()
response = client.get("path/to/resource")
athert(response.status_code).equals(200)
$ athena run ./my_module
my_module: failed
│ File "/home/haondt/projects/my-project/athena/my-workspace/collections/my-collection/run/my_module.py", line 8, in run
│ athert(response.status_code).equals(200)
│
│ AssertionError: expected `200` but found `404`
Setting up a Mock Server
athena includes a module that wraps Flask
to deploy a mock server. To use it, create a file anywhere in the athena
directory named server.py
.
This module must contain a method with 1 argument called serve
. This method will be given a ServerBuilder
as the argument.
from athena.server import ServerBuilder
def serve(builder: ServerBuilder):
builder.add_server(lambda c: c
.host('0.0.0.0')
.port(5000)
.get('/api/hello-world', lambda r: r
.status(200))
)
Configuring the server
Multiple servers can be implemented, but each server must be on a unique port. If the port
setting can be omitted from the builder, athena will automatically assign
an incremental port to the server.
from athena.server import ServerBuilder
def serve(builder: ServerBuilder):
builder.add_server(lambda c: c
.get('/api/hello-world', lambda r: r
.status(200))
)
builder.add_server(lambda c: c
.get('/api/hello-world', lambda r: r
.status(201))
)
The add_server
method accepts a lambda that will configure the endpoint using a ServerConfigurator
. The ServerConfigurator
can
be used to configure the server itself (host
, port
, etc) as well as adding routes using the get
, post
, send
, etc methods.
Configuring the route
There are several methods on the SeverConfigurator
to add a route using a RouteBuilder
. For more complex routes, the RouteBuilder
can be configured from a defined function.
from athena.server import ServerBuilder, RouteBuilder
def serve(builder: ServerBuilder):
builder.add_server(lambda c: c
.get('/api/hello-world', hello_world)
)
def hello_world(builder: RouteBuilder):
if 'X-API-KEY' not in builder.request.headers:
return builder.status(401)
if builder.request.headers['X-API-KEY'] != 'foobar':
return builder.status(401)
return builder.body.text('hello, world!')
The builder provides methods on the object itself for configuring the response, e.g. RouteBuilder.status
, as well as a
RouteBuilder.request
accessor to retrieve information about the incoming request,
e.g. RouteBuilder.request.json
.