Integration testing tests how different parts of the system, which in our case is the ELT pipeline,
work together.
An example of this would be the part where we transform the data from staging to the core layer and
upload it to the Postgres table.
So here we would want to test that the transform data, which we create using the Python functions is
updated correctly to the Postgres database.
As we said in the unit test lecture, we will aim to use real credentials for integration testing.
And we can start off with what we tested in the unit tests with the airflow variables and database connection,
but using real credentials.
Let's start with the variables part by first defining a function.
Get underscore airflow underscore variable in the test script.
Since this function will be reused.
This function will return the value of the variables from the environment by specifying the variable
name as an argument.
We then construct a full environment variable name by defining a variable env underscore var and set
it to the airflow variable.
So what this piece of code over here does is it gets the string variable defined and makes sure it is
all in caps.
And finally we are retrieving the value from the environment.
Now to use this helper function as a fixture what we need to do is to wrap it inside another function
using the Pytest fixture decorator.
So the function we just defined will be inside this outside function airflow underscore variable, which
is wrapped around the pytest fixture decorator.
Finally, we simply return it.
So return.
We can now create a test that uses these variables.
A simple test would be to use them as input variables for the API connection, to see if we get an okay
response or no.
So let's first create another script which we will call integration underscore test.
Py.
And here we will build a function.
We will call it as follows.
Here we need to make sure that we pass the airflow variable fixture as an argument.
We can then get the API key and the channel handle using the airflow variable fixture which we just
built.
And as you can see here we are defining it in both the API key and the channel handle.
Now we can define the URL.
So this URL is actually from the very first function which we built in this course which was the get
playlist id function.
Now to make the API request it would be better to use the try and accept clauses, since we are connecting
to the actual real database, as there are more potential fail points than when using a mock connection,
such as network issues, timeouts, authentication failures, to name a few.
So let's define the try except block.
And here we are making the API request using the request method.
So to do that we actually need to import the request module.
Another thing that we will need is the Pytest module.
Let's import Pytest.
What is left is to assert if the response status code is a successful one, which we know is 200.
So let's just say this right now.
So to give a brief summary, what we're doing here is we're checking if the response that we get from
the URL where we are using the real channel handle and the API key will give a successful response.
If not, then we define in the accept block the pi test fail function, which causes the Pi test to
mark the test as failed, and we can also specify a failure reason in the Pi test output.
So that's it for this function.
And now we can actually test it out.
So we can use the same structure of the Pi test.
Testing syntax.
What we need to do is to change the actual function.
And here we replace it with the functions we just built.
So actually I'm seeing that we're deselected.
None were actually run.
The reason being that we are no longer using the unit test script, but we are instead using the integration.
Underscore test pi.
So this should turn the result now.
And in fact it did.
And the test has successfully passed.
So what we're saying here is that the response was actually a 200 response code.
Okay.
So at this point we have just used the real API key and channel handled variables that will be used
in the actual deployment of this ELT pipeline.
Another test that we can do is with the connection to the database that will store the ELT data.
Since this connection can be used by other functions, we again go to the conf test script and start
writing the fixture that will contain the connection with the real credentials.
As we have seen so far, we use the Pytest fixture decorator and define an appropriate function name.
In this case it will be real Postgres connection.
Then, as we have seen already, we retrieve the connection details to the Postgres database from the
airflow environment variables using the Os.getenv function where we have the following variables.
Next we establish the connection to the Postgres database using the Psycopg2 library.
So let's go at the very top and import the library here.
And since this is a real connection we use, try and accept the block again and start passing the connection
parameters, same as we have done in the database section of this course.
We define a connection variable.
Let's name it and set it to the value of none.
And since this is already a connection, we use a try and except block and start passing the connection
parameters.
We can't forget to provide the connection to test, so we yield the connection.
Then we handle the errors in the except block.
And finally we ensure that the connection is always closed.
So finally if connection is defined connection dot close we can come back to the integration test Script
and start writing our main function to verify the poses connection using the real credentials.
Same as we have done with the other functions, we define the function that uses the fixture from Conftest.
We then define the cursor variable as null, which will help with the closing logic in the finally block.
Using the try and Accept and finally block, we can start off with the try block and define the cursor
using the following code to verify that both the database connection and the SQL execution pipeline
are working fine.
We use a basic Select one SQL statement that selects the constant value one.
So when we come to fetch the first row of the result set using the fetch one method, it should return
the value of one.
We then do this section by writing insert result zero equals one.
We have covered this already, but to be sure everyone is on the same page, we specify the first element
in the result, which is this one.
Here, because the result is a tuple when created using the cursor dot, fetch one method.
So it would return a tuple as follows.
So in summary, to access the first and only element, we are using the indexing.
So here we can remove this.
And also here we terminate the try block.
Next we can move on to the except block.
But first we will need to do an import of the Psycopg2 module.
So coming back to the except block we state what happens when there is an error, which we do by defining
the cycle dot error class and pass the exception to the variable e.
We use the Python function, which will mark the test as a fail and return the text in the quotes with
the exception.
Finally, we use the finally block similar to what we did for the connection.
This time to close the carousel.
Like that, the function is complete and now we can run the Pytest test.
As you can see, the test has outputted a pass.
So this means that the credentials that we are using to connect to the database are working as expected.
And the select one statement, the SQL select statement indeed ran and we get the result as expected.
With that, we have concluded our integration tests.
There is much more you can do, and I invite you to create integration tests of your own to get more
practice.
Next up we will look into the end to end tests.