This code is for tutorial purposes of the Agile Actors Data Chapter. It continues in a more advanced manner from Basic integration Testing in the spirirt of Testing in Airflow Part 2 — Integration Tests and End-To-End Pipeline Tests

In summary the project sets up an SQL Server with "transactions". Airflow scans the table and when it finds new transactions, it ingests the transactions in S3Ninja emulating S3 store. Then it marks in another table that the task has been accomplished.

Tested with python 3.11.3.

First install dependencies

pip install -r requirements-dev.txt

and then build "good" image.

docker compose build

Now testing is as easy as executing

 rm -rf logs s3data s3logs
 pytest -vvv -s --log-cli-level=DEBUG tests/integration/test_automatically_sample_dag.py

The integration tests (not complete) create the bucket in docker-compose.yaml, the test database, user, schema and table also in docker-compose.yaml. Then it runs tha dag.

Here we run manually the steps leading to the integration tests. For maintaining state uncomment the sections in docker compose for volumes, not recommended. Please comment them again before running the integration tests.

Start the deployment by running

rm -rf logs s3data s3logs
docker compose up

Now you need to what is in docker-compose.yaml.

Connect to S3Ninja and create a bucket named mybucket. This is the name that appears in airflow variable.

Connect with DBeaver to the SQL Server with credentials

user = 'sa'
password = 'newcross_123'
host = '127.0.0.1'
port = 1433
database = 'master'

Now for cleaning everything up (if you need it), just run teardown_database.sql

To create the database used in the deployment execute

IF NOT EXISTS (SELECT * FROM sys.databases WHERE name = N'ncintegration')
    BEGIN
        CREATE DATABASE ncintegration;
    END

and to create the rest, just run setup_database.sql

You can optionally for your tests add

delete from ncproject.fintransacts;
insert into  ncproject.fintransacts values (1, convert(DATETIME, '1968-10-23 12:45:37', 20), 'buy 1');
insert into  ncproject.fintransacts values (2, convert(DATETIME, '1968-10-24 12:45:37', 20), 'buy 2');
insert into  ncproject.fintransacts values (3, convert(DATETIME, '1968-10-25 12:45:37', 20), 'buy 3');
insert into  ncproject.fintransacts values (4, convert(DATETIME, '1968-10-26 12:45:37', 20), 'buy 4');
insert into  ncproject.fintransacts values (5, convert(DATETIME, '1968-10-27 12:45:37', 20), 'buy 5');

Visit the airflow server and activate your dag. Extra info in official Airflow Documentation. Airflow Apache Project.

Have fun!