foreign
welcome back to our last talk before
lunch I'm very excited to Welcome to the
stage Jenny who is a software engineer
at car Calabria and who is also an
Outreach alumni of the December 2021
cohort I'm incredibly excited to be
hearing about testing asynchronous
applications with fast API and Pi test
Jenny take it away thank you
good afternoon all uh I'm Jenny and I'm
going to talk about testing asynchronous
applications with fast API and Pi test
today so before digging into the topic I
would like to briefly introduce myself
so I come from India and I work as a
software engineer and an open source
consultancy called collabra before
joining collabora I was announced your
intern so that's where I started my open
source journey I was working in a
conversa organization
um
and
um I kept contributing to that project
while being in collabora so now I'm a
co-maintainer of one of the projects uh
if I talk about my technical interest of
course Python and Linux fascinates me a
lot and apart from that I love dancing
and I never miss a chance to dance if
the organizers would have given me extra
few minutes you would definitely see me
around dancing on the stage
uh so let's see uh these are the agendas
of today's talk I'm going to present
what is first API a brief introduction
about what is asynchronous programming
then we're gonna see uh how to write
test for the synchronous code uh then
I'm going to present a very interesting
test examples uh from Kenosha API and
that is a back-end from kernels sky so
before going uh into that I would uh
briefly introduce what the project is
about
and then um yeah we'll see something uh
some fast API in the practice
so okay uh first of all what is fast API
um as we all know python provides a
variety of web Frameworks including
Django and flask so fast API is one of
the framework and what is special thing
about is is it is the statistics in
Benchmark shows that it is one of the
fastest Frameworks available
first API supports python 3.7 version
and newer than that
first API is based on Starlet and by
dantec those are two different packages
so uh for the people who don't know our
Starlet is a package that is specially
designed for writing the asynchronous
web services and you can see that first
DPR is a subclass of Starlet so it
directly supports all the features that
Starlet provides for example websockets
bought then background tasks then events
even handlers like startup and shutdown
events
it also has some inbuilt security
utilities and dependency interactions as
well now what is patient python Tech is
a data validation package provided by
python it works uh based on type ends so
it has a by default validators and it
can be useful to validate your request
and response parameters so it is pretty
useful you don't need to do everything
manually
apart from that first API provides
automatic interactive documentation it
is based on open EI and you can have
either speaker paste documentation or
retalk generated documentation it is
pretty interactive so you that you can
test your client application from the
browser itself so these were the
features of fast API now we'll see what
is the synchronous programming
so um programming Paradigm can be a
defined with the two kind of Paradigm
one is the synchronous and another one
is asynchronous so when multiple
processes are there and all the
processes are running sequentially like
one by one that is called the
synchronous programming right and when
multiple processes are there and they
can run at the same time of course not
like it is not necessary that they are
running simultaneously but if they are
running in overlapping manner as well so
you can see that that is asynchronous
programming and what is the special
thing about asynchronous programming is
it has the ability to wait for a task to
complete and do something else in the
meantime so we'll explain this in a very
uh simple example
so I have um attached to pictures here
the first thing the first picture shows
that a very long queue of the people so
imagine that you are visiting a museum
in Adelaide you have already bought a
ticket now you are in a queue and you're
waiting for your turn to come up to
enter the museum
so um if you imagine that every single
people in the queue is every single
process then they can execute just after
the prior one completes like for example
if the first one first person goes in
then and then the only the second person
can goes in so that is the example of
synchronous programming and while
waiting in the queue they cannot do
anything else but just waiting right so
that is an example of synchronous now
another picture is about restaurant so
there are multiple tables and multiple
group of people are there
if you imagine every single group of
people as a separate process they can
order food at like at the same time of
course not simultilies but they can do
that at the same time and while waiting
for the food to get delivered they can
have a chat with each other or scroll
their phone or do anything else they
want so that is the ability of
asynchronous code so that they can do
like simultaneous thing well we well
they are waiting for a particular thing
to happen
okay
now uh how to write asynchronous code
with a fast API so first API is based on
any i o package the Nar package is in a
tool or you can see framework to write
asynchronous networking apps so fast API
supports that first API also supports
this two backends called async IO and
Trio packages so you can use any of this
package to write your asynchronous code
with a fast API moreover python has the
modern versions of python has introduced
functions called Cod routines so it is
just a fancy word about the asynchronous
function so when you have async and
await keywords with your functions then
you can say that this is the
asynchronous functions
here is a pseudo code so I'm trying to
get some documents from my database so I
have defined that function as
asynchronous so I've done that by using
this keyword that is async Def so it
will know that this is the asynchronous
function and um while the database go
and fetches documents from the database
I will wait to get the documents so I
have done that by using the await
keyboard so await will wait for the
response to get right so and I'm simply
returning the documents from the
function so this is a very basic suit
example of finishing Chrome is code that
you can write for the first API
uh now I'm going to discuss like few
bullet points about how I write a test
for the asynchronous code this is not
like official documentation or any
standard things but this is the uh
somebody that I find helpful for myself
so first
um if you are writing a test case then
you will need a test client so for the
synchronous test cases it provides a
test client called it it by default it
has the test client so you can use for
the first API for the asynchronous we
will be using https library that is the
another library for HTTP handlers from
python so you can use asynchronous test
plan from that Library
after that we will be going to use some
of the pi test fixtures so when you are
writing a test case for some kind of a
specific scenario that that will be case
that you will need to have some data fed
into your test case for example if you
need to initialize your application if
you want to setting things up if you
want to set your database or any kind of
service setup then you will need those
kind of information before you actually
run your test so those kind of functions
and utilities you can write as a pie
test fixtures and then you can use it
directly from your byte from the from a
test case
okay uh if I am writing a unit test if
my application has multiple components
and multiple modules in my application
so in the unit test what we do is we
just focus on a single separate
component and we write test cases for
that if I have application and some of
these API services that I'm using for
example if I'm using database service
rate is service or any other kind of
service but I just want to focus on a
single component API then if my
application is communicating with this
other modules then I need to mock some
function calls when I'm writing unit
test because we are just focusing on
single module and will be not having
those Services up and running so the
communication won't be able to happen
so in this kind of scenarios we need to
mock function calls where we require
then of course actually uh write your
test function the actual test function
then you need to Define what the test
what kind of test environment uh you
would like to run and then of course run
your test
okay uh so I'm going to uh describe some
of the test examples from karnesia API
but before that I would like to
introduce this project uh this is called
karna CI so as we all know Linux kernel
is running on variety of Hardwares and
devices right so it will be very
difficult to maintain its stability and
quality around this range of Hardware so
in 2014 a Linux kernel a few Linux
kernel developers and a few uh community
members from Community decided to have a
specific CS system for Linux kernel
that's where this project got started
and now it is a Linux Foundation project
so basically Linux kernel uh Kenosha is
in shorting the quality stability and
long-term maintenance of Linux kernel
since simpler was what you can do with
Garcia is if you have your specific
Linux kernel then you can build your
Linux kernel with Canon CI tools then
you can have a set of tests on the on
your kernel and then you can also submit
your test to a common database tool and
that will be accessible by publicly
so you could have your test reports
regression reports even by section
reports with Gunner CI if you are more
interested then I have mentioned a link
there kenosha.org you can go and check
out the project in detail and the reason
I'm discussing kernel CI here is a
conversion has a new backend called
karnesia API and kernelsia API is based
on these uh major building blocks it it
um it is built on top of fast API so
that was the reason I'm discussing this
project for the database it uses mongodb
and that it is an in-memory database
and it also has a storage support you
can use as SSH for the upload and HTTP
for download so that uh this I'm not
going into detail because this is out of
scope for this uh talk but everything is
up um publicly on GitHub if you want to
access this project you can go with this
you can go on this get up link and find
API project
now let's dig into the text examples so
first I am going to describe the unit
Test example
um so
API has a root Handler a root Handler of
course we all know that that is the
entry point or initialization point for
any application so in case of currency
API it returns this dictionary uh with a
message as a queue and currency API is
as a value it is a pretty uh basic
endpoint the root and the one so if you
send the HTTP get request to this root
endpoint it will be uh this it will be
providing this response
now how to write test of this kind of
endpoint so as we saw uh as we saw that
we need to have some kind of test client
to write the test for it so I have used
uh fast api.test client package to use
the synchronous test client so if you
see the first frame I have shown in the
snapshot is I am creating an instance of
the test client it will need two
arguments one is a application instance
and another one is URL of API server so
in the unit test we are not going to run
any kind of server the actual server so
what we can do is we can use the demo
server or fake server provided by fast
API so this is the default server that
is HTTP test server and I'm going to use
it in my test case
please note that I have written it as a
form of Pi test fixture because test
client is something that will be knit
needed to fit with every test cases that
I write okay so I would need that for
the initialization so I have chosen it
to uh to be as a pi test fixture
and another function is test root
endpoint this is the actual test case of
my unit test so what I'm doing is I'm
passing the test client the fixture name
in the argument so that is the way of
using pi test fixture you just uh need
to pass it in the argument and it will
be executed before actual test case
executes
so what I'm doing is I am sending HTTP
get request using that test client with
test client.get method and I'll be
receiving some HTTP response we can use
uh some asset statements to verify what
your test has received as a response so
you can I I have used this asset the the
last frame of the snapshot that I said
response dot status code equal to 200 so
if this condition fails it will raise
the session and a error and my test will
fail
okay so let's try to run a test of
course there are variety of testing
environments that you can use but I have
chosen the very basic one and this is
the command line utility so I'm running
Pi test commands on the command line to
run my test so let's see how it goes
oh and the test got failed why so in the
snapshot I have shared that these These
are the error logs that I've received
and it shows that there is some some
kind of startup Handler is there and it
is trying to connect one method called
Pub sub.create and again uh the last
line of the frame shows that it is
trying to connect with some method
called Pub subtitude in its sub ID
so under the hood what happened is Now
API is uh has published a subscriber
mechanism implemented so it is using
some kind of tool called redis so red is
a in-memory database uh tool so so you
can use it for your pop sub mechanisms
so under the hood karna CI API is trying
to connect with the treaty server on its
initialization but it is not trying to
connect like it is trying to money but
it got failed of course this is expected
outcome because we are not running any
redness server with our test right
so how do you deal with this kind of
situation so I've also mentioned that
you you can mock function calls whenever
you need it so what you will do is
you'll need you'll have a mock function
called to your method called Pub sub.net
sub ID so there is a package called Unit
unit test from the python
and if you are writing tests for the
asynchronous functions and you want to
mock that function call you you can use
the specific async mock from the unit
test package
okay and uh though this the first frame
I have drawn here is
it is uh it it has created one async
mock object and the another line says
that mocker.patch it is used to patch
your class level methods or even modules
to patch that particular function with
the mock object that you have created so
with the async mock I have mocked the
function call that is popsup.net ID so
what now uh will happen is when uh when
the pop sub dot init sub ID method gets
called instead of calling that specific
method it will go straight to the
asynchronous mock object so nothing will
break in this case so I have defined
this as a pi test fixture as well so I
am passing the name of the fixture in in
my test case
that is just root endpoint and I'm
passing that function name so um so
whenever it tries to connect the pop sub
method it will uh go to the asynchronous
mock instead right and then uh the rest
is same I have not updated anything in
that code so let's try to run this test
again
so now yes it passed so we have used uh
by test fixtures and we also tried to
Mark some function calls with this test
so that was uh the example of unit test
now let's see an example of end to end
test so before that let's see what this
end-to-end test is for the unit test we
are we were trying to test a single
component or single module at the same
time but for the end-to-end test when
you are testing your software
implication from a specific workflow
from the start to end so that is called
the end-to-end testing
for the end-to-end testing we you need
to have a well-defined workflow before
you can test it so uh so I have
demonstrated a very minimal pipeline
here
so I've already said that because API
has published a subscriber mechanism in
place so I'm I'm going to try and use
that interface to test my whole pipeline
here
so what I'm going to try here is for I I
will be writing a subscriber subscriber
mechanism and that will be subscribed to
a specific Channel called the user
channel so the user channel will receive
all kinds of events when uh user
accounts will be created or updated or
even removed so these kind of messages
will be broadcasted on this channel
so we'll be subscribing to that channel
then we'll be listening on that channel
rather I'd say that we'll be polling out
the channel to receive the continuous
events so that will be the infinite Loop
and meanwhile a user account has will be
created so what I expect is when a user
account gets created the created event
should be broadcasted that channel and
the subscriber should receive it right
so we'll we'll check if that happens or
not and at the end of the workflow I am
going to unsubscribe the channel
okay
so uh how do we write test cases for it
as this is the pipeline that I've shown
it is completely asynchronous all the
functions of that pipeline is going to
be asynchronous and that's why we we
need to use the synchronous test graph
right
so the httpx library provides the
asynchronous or test client and we are
going to use it so I have another I have
defined another Pi text fixtures that is
called test async client and what I'm
doing is I'm creating an instance of
that asynchronous test client
so it is same as we did for the
synchronous test the difference is
instead of using the fake test server
URL as as your base URL you'll be
needing an actual API server URL for
example if my API is running on this URL
called HTTP API Port is 8000 and the
version prefix is latest so if my API is
running on this particular survey URL
then I will need to use that for my
asynchronous test client because at the
end I'm going to run my EPS server and
it will be trying to connect with that
actual web API server
so that is the difference between
synchronous and asynchronous test
the next thing is we will be subscribing
to a channel called user so API has an
endpoint defined to do so
so what I'm doing is I am defining a
test function called test subscribe I'm
using the fixture for the testing
asynchronous client so so I have passed
the fixture name in the argument that is
test async client and please note the
first frame of this snapshot that I have
used two decorators here so first one is
pi test.mark.isync articulator so if you
are writing a test for some kind of
asynchronous function then you need to
Pi test uh then you need to let the
pythons know that this is asynchronous
test and not synchronous test so you can
do that by using this decorator called
pie test dot Market async IO another one
is
bytes.mark.edu order so when you have
multiple tests in your uh testing for a
specific pipeline or a workflow you will
need that some kind of test function
should be executed at the very beginning
of your flow for example if subscription
is not going to happen we we are not
going to unsubscribe anything right so
we we need that this test subscribe
function should be executed at the very
beginning so I have a marked order and
I'm passing the value one so it is the
way of saying that this test should be
executed first
and what I'm doing in the test case the
test subscribe subscribe uh test is I'm
sending the HTTP post request to this
endpoint subscribe slash user endpoint
and what I'm receiving is I'm receiving
a subscription ID as HTTP response so
now see the last uh frame of the
snapshot
a pie test has a way of sharing Global
variables so if I need uh some kind of
information and that should be shared
among all the modules or all the test
cases then I can use the pytus global
variables and this is the way of
defining it so if I write Pi test or
subscription ID so so that subscription
ID variable will be accessible through
all your test cases
so now we are going to listen to that
subscription ID as I've already said
listening is a kind of in a infinite
look so it should be run as a background
task rather than it rather than a
traditional asynchronous test function
right so a way of doing is uh there is a
function there is a package called async
IO and it provides something called task
task objects so you can use that to
create uh to create your task that
should be run in the background
so here I'm going to listen on the
subscription ID and I'm uh and I'm going
to create that task for uh so that would
be that would be run the background task
because this is the way of defining a
background task in asynchronous code so
I have used asyncio.create method and
I'm passing the endpoint HTTP get
endpoint to that Handler so it will
return me the task Handler and from the
time to time I can also track the status
of that Handler
so now this is the main function of this
whole workflow I've defined a test case
uh called test Pipeline and please note
that this uses The Decorator called Pi
testoc mat.dependency so what it means
is this if some kind of test cases
depends on variety of test cases you can
Define the dependency like this so it
shows that this test function uh depends
on test subscribe so if the test
subscribe gets failed or anything happen
to that test case this test case will
simply be skipped
so that is the utility of Depends
decorator
now in the test function I am creating
one listen task so I'm polling uh for
the different kind of listen events uh
listen events by by using this task
Handler and it is returning task listen
task
and meanwhile what I'm doing is I'm
creating a user account so I'm sending
HTTP post request to the user Handler
and I'm passing very basic information
that is username and password to just
create a user account so what I expect
is the task listener should receive an
event that an user account has been
created so if if the status has been
updated for the task Handler I can get
the status using the await keyword so
what I've done is
I've used a weight and then my task
Handler so if sorry so if I get the
event data from my task Handler and I I
also can verify my event data that if I
intend to receive the uh user a user
creation event then I should have
received it right so I've used a bunch
of assate statements to verify that
so I have uh that event data should have
like two parameters one is the operation
and another one is the ID so it shows me
that operation is created means user
account has been created and ID will
show that what is the idea of the user
user account that has just been created
so you can use bunch of acid statement
to verify that cloud event as well
and at the end of the flow I am
unsubscribing to the channel
so um I have uh used another decorator
stock market order and have passed value
last so when you want some function to
be executed at the very end of your flow
then you can do that by providing last
as a value to Your Pie test order
decorator so I've done it so unsubscribe
is something that will be executed at
the very very end of the flow
and now we have the actual test function
called test unsubscribe and what I'm
doing is I'm just sending HTTP post
request to the to the unsubscribe
endpoint and again I'm using that pie
test or subscription ID variable here so
I have shared that variable among all
the test cases and I've used at so many
places so that is uh the reason that I
have declared it as a global variable
and I'm at the end I am just verifying
that if the HTTP response code is 200 or
not
so that was the all the test cases of
endpoint is now it's time to run all the
tests
so for this I need to actually run my
API server and then I would need to run
my Reddit server as well because for the
publisher subscriber mechanism I have
used a Reddit server so I've done that
and after starting my all the servers as
I'm running the test from the terminal
itself
so it shows that all the test cases that
I have run from my end-to-end test so
you can see that test subscribe executed
at the very beginning because I I pass
the order value as 1 for that test case
and another one is the test plan test
Pipeline and at the end it uh it runs
the test unsubscribe method
so uh the pi test function also gives
you the status and uh yeah it also shows
the Temptation that it took to run your
oil test
so because of the time constraint I just
uh could explain two two test cases that
is unit test and end to end test if you
are interested in if you want to check
some of the more examples from the
condenser API that I have put the GitHub
link and you can go and check that
so that result that was all for me thank
you so much
[Applause]
thank you very much here is a small gift
as a thank you for giving a session
thank you and for 10 days that haven't
been able to see this it is a mug with
this year's 2023 artwork on it so it's
the exclusive speaker mode
congratulations thank you thank you so
much
ah
we