The main aim of unit tests is to test out individual parts of the code in isolation, to ensure that
it works correctly.
When we talk about individual parts, we are referring to the Python functions we have created in this
course.
A common convention for unit tests is to use mock credentials.
This is the standard approach because you want to isolate the unit being tested from external dependencies
like databases or APIs.
For integration tests, you can use mock or credentials depending on specific scenario.
However, in our case we will use real credentials and for the end to end tests, it is necessary to
use the credentials to simulate the actual process and in our case, ensure that the ELT process is
working as expected.
When performing tests using the Pytest framework, you should be aware of the conf test.py file, which
we will create soon.
In this script, we will put the reusable code that will provide inputs for the test functions, which
we will create in this script.
We will have common setup code like database connections, API keys and more.
This reusable setup code is known as fixtures, and to create such fixture we will use the Pytest fixture
decorator.
I will leave a link to the Python documentation that explains in more detail what the test is.
One last important thing we should know is that to use the Pytest framework when creating the testing
functions to enable the Pytest testing, we will need to define the names of the functions to start
with test underscore.
Here you see such function where the first part is test underscore.
So now that we have covered the basics, let's create a Python script under the test directory and give
it a name unit test.
Here we also create the script which we mentioned earlier.
Let's start off with our first unit test where we will mock the API key.
Since the API key is something that can and will be reused by other functions, we can define the code
for this function.
In the test, we first need to import some essential libraries.
These are the OS pi test, the mock module from the unit test library and variable from the airflow
module library.
And now we can start off by using the Pytest fixture decorator, which as we have already said in Pytest.
This indicates that this function is a fixture and will be used to provide input data for the main test
functions.
The syntax is as follows.
We use the add for decorator and that's how you define the syntax.
Then we define the actual function.
Let's call it API underscore key in our case.
Next we use the mock dot Patch dot dict, which handles dictionaries to mock the environment variable.
Airflow.
Underscore.
Underscore API.
Underscore key.
What this piece of code does is it temporarily updates the environment dictionary with the key value
pair that we have here.
Finally, we use the yield to provide the value to the test that requests this fixture.
As we saw in the airflow section, variable dot get fetches the value of any airflow variable.
And in this case we want the API underscore key variable.
Going back to the unit test Python script, first we create the function which as we said, needs to
start with test underscore and then give it an appropriate name which is API underscore key.
This function will need to take as argument the API underscore key fixture we defined in the conf test.py.
So let's start by writing test underscore API underscore key which takes API underscore key argument.
Then to verify the API key, we simply use the Python assert keyword to test if API underscore key is
in fact equal to what we defined in the fixture, which should be mock underscore key 1234.
Before we check if the test passes or fails.
A quick note that I want to add is that apart from naming conventions for functions, Pytest also has
naming conventions for the script names.
So we name the script unit underscore test because Pytest requires that the Python script starts with
test underscore or ends in underscore test.
And here we are using the latter letter syntax.
So going back to the results of our test.
Let's go inside one of the airflow containers.
Here we already are inside the airflow worker container.
And let's run our test.
So we write Pytest dash v.
We use the dash v for a more verbose output, and then we define the path of where the unit test script
is, which is tests unit underscore test py for now.
To run this script, it would give the output for the single function.
However, we will continue to build more tests.
So to specify a particular function we use the dash k option with the function name.
Let's run it.
So the idea is that we named the directory tests and not test.
So let's go back here.
Make sure that this is saved both here and here.
And let's run this again.
And there we have it.
Our test has passed.
We can do the same exact thing for the channel handle, which, as you might remember, is Mr. Beast.
We just need to change the function name and use the appropriate variable name in airflow, so we can
copy and paste the previous function and the y and replace the appropriate variables for the channel
handle.
Again, we can set the channel handle to what we want, even something silly like Mr. Cheese.
So let's go back to the unit test script.
We copy this.
We do our changes.
So now that all the changes have been done, let's run this test again.
We should also get a passed.
Notice how now, since we have two tests and we only specify the channel.
Handle test.
The API key test would be shown as the selected.
Now that we have created unit tests for variable mocking, let's do the same for database connections.
Specifically the database connection to where we are storing the API data.
And in order to do this we need to import the connection module from the airflow dot models.
So after variable we can add connection.
We can also create the fixture with the function name as before.
Changing this to our next target function.
To create our mock Postgres connection, we can define a mock connection variable named Conn and set
it equal to the connection class objects, which is composed of the login password, host port and the
schema.
The schema being the database name and an airflow connection class.
We can then use the get underscore URI methods to create the connection URI for this Postgres connection.
And finally use the mock dot patch dot dict function.
Same as before, but this time we are mocking the connection URI.
So we use airflow underscore conn and then the connection id.
So it comes to as follows.
Finally, for the main function to get the connection, you can use the get connection from secrets
methods provided by airflow to fetch the connection details stored in Airflow Secrets backend.
Here we just need to specify the connection ID, which is the Postgres underscore db underscore YouTube
underscore elt.
Now all that is left is to go back to the unit, test Python script and write our assessment statements,
same as we did for the variables.
So we define the function name with test underscore at the beginning.
So we can write test underscore Postgres underscore one.
And we need as argument the mock Postgres converse fixture which we define in conftest.
Here we simply assign the Postgres connection provided by the test fixture to a variable conn to make
the assertions are easier to read and write, and from here we simply define the assertions as follows.
Unlike that, our function is complete.
To test this out using Pytest, we can run the same command that we have used for the previous two tests.
Obviously specifying the function name that we just created.
Press enter.
And as you can see, all the tests are passed.
One final unit test that we can do involves testing with our DAGs are structured as we expect them to
interact with the DAGs.
We use the Dag bag instance, which collects all our DAGs information so we can continue the airflow
of models.
Import by adding Dag bag.
Now that we have the relevant import in the test script, we can create a very simple Pytest fixture
that returns the Dag, which will be as follows.
So that's all there is to it for the conf test script.
Now let's add our focus to the unit test script and start writing our code.
We first define the function and as an argument use the fixture we just created.
Now this test, as we said, will test the integrity of our Dex.
We can split this integrity testing into four parts.
The first one being that there will be no import errors.
Then the border Dex we defined are being loaded.
The the number of Dex is correct.
And then finally each Dag has a number of tasks we expect.
Let's start off with the no import errors check.
And in this test we will also include print statements.
So when we come to run the Pytest command we can actually see what is being tested under the hood.
This is a simple assertion where we assert that the import Errors instance from the instance is equal
to an empty dictionary.
This means that we have no import errors.
If, however, our assertions fail, we can use the assert statement to return a descriptive message
showing the errors.
As we said, we will print the object as a way of debugging our code, but also seeing what is happening
under the hood.
The printers we have here is simply to separate the prints that we will have from all of these test
integrity functions.
The next subtest is to test that all expected DAGs are loaded.
So first we can write all the expected Dag IDs in a list.
Then we get the actual Dag IDs from the Dag keys using this command.
And then we can print these out.
Remember that these Dag keys will be the decades we defined in the Main.py script under the DAGs directory.
Now using a for loop, we go through each tag ID in our expected Dag IDs list, asserting that this
Dag id is indeed found in the Dag keys.
Before we move on to the third subtest, let me just split this so it's more readable.
The third test is a very simple count of DAGs.
We know that so far we have three DAGs.
So this assertion should return three.
We therefore write the following.
And also we need to print out what we are asserting.
Finally we test that all loaded DAGs have the expected number of tasks.
So first we define a dictionary with key value pairs being the dag id and the expected task count for
each Dag.
So just to make sure we're all on the same page, the produce underscore JSON, we expect it to have
four tasks.
So if we go in main.py for the produce underscore JSON that id, we expect that we will find four tasks.
Next we print the separator.
And here we define the for loop that goes through each Dag.
We define the expected count variable from the dictionary we just constructed.
And the actual count variable which we create by getting the number of tasks for the Dag using the length
function.
We finally do an assertion, and if this fails, we can print out a command as follows.
For debugging sake.
We finally print out the Dag underscore ID and the number of tasks for each Dag.
So this test encapsulates a number of subtests.
If we had to run it using pytest.
Using the command we have been using so far, it would work in telling us if it passed or failed, but
we wouldn't get the prints we defined.
This is because by default Pytest does output capturing, meaning that outputs from prints are by default
not shown.
To disable this option, we can use the Dash s flag so we can write the following.
So after V, we can write dash s and we simply change the function name here.
So as you can see, all of these subtests that we have have passed.
So you might say, how would it look like if one of our subtests have failed?
Let's create an erroneous assertion and say that the produce JSON, the ID has only three tasks.
Let's run this test again.
And as you can see, the assertion error here is telling you exactly how many tasks the produce underscore
JSON has.
Here we are saying that the test three.
But in fact it actually has four tasks.
So this assertion will fail.
So now I will simply change this back to four.
So at this point we have went over some basic but important unit tests.
Feel free to add more tests to keep on practicing.
Next we will move on to the integration testing.
I will see you then.