Overwriview

• 🛠 What is TDD
• 📝 TDD for the new architecture
• 🎨 How to do unit test
• 🧑‍💻 Hot to do integeration test

What is TDD

Test Driven Development (TDD) is software development approach in which test cases are developed to specify and validate what the code will do. In simple terms, test cases for each functionality are created and tested first and if the test fails then the new code is written in order to pass the test and making code simple and bug-free.

• Add a test.
• Run all tests and see if any new test fails.
• Write some code.
• Run tests and Refactor code.
• Repeat.

What is TDD

TDD usually follows the "Red-Green-Refactor" cycle:

• (Red) Run all the tests to ensure the new test fails
• (Green) Write just enough code to get that single test to pass
• (Refactor) Improve the initial code while keeping the tests green

Pytest

The pytest framework makes it easy to write small, readable tests, and can scale to support complex functional testing for applications and libraries

• Easy to learn: Tests using pytest are Python functions with test_ prepended or _test appended to the function name
• Test filtering py.test tests_directory/foo.py tests_directory/bar.py -k test_001 or test_some_other_test
• Fixtures: Fixtures parametrization is another feature that help us send parameters used during fixtures execution

Unit Test

A unit test is a way of testing a unit - the smallest piece of code that can be logically isolated in a system. In most programming languages, that is a function, a subroutine, a method or property.

Why perform Unit Testing?

• Unit tests help to fix bugs early in the development cycle and save costs.
• It helps the developers to understand the testing code base and enables them to make changes quickly
• Good unit tests serve as project documentation
• Unit tests help with code re-use. Migrate both your code and your tests to your new project. Tweak the code until the tests run again.

How To write Unit test

A unit test is a way of testing a unit - the smallest piece of code that can be logically isolated in a system

• Prepping and reviewing the unit in question
• Makeing test cases and scripts
• Testing the code

Unit Test Example

We need to add the user_service’s URL to setting , let’s write the test case first create a file in tests/unit/test_settings.py

from app.settings import settings

def test_user_service_url_in_settings():
    assert settings.USER_SERVICES_URL
from app.settings import settings

def test_user_service_url_in_settings():
    assert settings.USER_SERVICES_URL

Unit Test Example

Add the missing settings in app/settings.py

from horizons.settings.base_settings import EnvBaseSettings
from functools import lru_cache
from pydantic import HttpUrl


class Settings(EnvBaseSettings):
    PROJECT_NAME: str = "Sample Service"
    SECRET_KEY: str
    USER_SERVICES_URL: HttpUrl = None

    APPLICATION_MODULES: list[str] = [
        "app.auth.models",
        "app.item.models",
    ]
from horizons.settings.base_settings import EnvBaseSettings
from functools import lru_cache
from pydantic import HttpUrl


class Settings(EnvBaseSettings):
    PROJECT_NAME: str = "Sample Service"
    SECRET_KEY: str
    USER_SERVICES_URL: HttpUrl = None

    APPLICATION_MODULES: list[str] = [
        "app.auth.models",
        "app.item.models",
    ]

Unit Test Example

Add the missing config in app/.env

SERVER_NAME="sample"
SERVER_HOST="0.0.0.0"
SERVER_PORT=8000
SECRET_KEY="09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
BACKEND_CORS_ORIGINS=["http://localhost:8000"]
LOG_LEVEL="INFO"
DATABASE_URL="mysql://horizons:Horizons@2022@localhost:3306/sample"
TEST_DATABASE_URL="mysql://horizons:Horizons@2022@localhost:3306/sample_test"
USER_SERVICES_URL="http://api.user.joinhorizons.com"
SERVER_NAME="sample"
SERVER_HOST="0.0.0.0"
SERVER_PORT=8000
SECRET_KEY="09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
BACKEND_CORS_ORIGINS=["http://localhost:8000"]
LOG_LEVEL="INFO"
DATABASE_URL="mysql://horizons:Horizons@2022@localhost:3306/sample"
TEST_DATABASE_URL="mysql://horizons:Horizons@2022@localhost:3306/sample_test"
USER_SERVICES_URL="http://api.user.joinhorizons.com"

Unit Test Example

Run the test again

poetry run pytest tests/unit/test_settings.py -k 'test_user_service_url_in_settings'
poetry run pytest tests/unit/test_settings.py -k 'test_user_service_url_in_settings'

we are great to see it’s success

Integration Test

Integration testing – also known as integration and testing (I&T) – is a type of software testing in which the different units, modules or components of a software application are tested as a combined entity

Why perform Integration Testing?

• It effectively executes important user-based scenarios.
• It might uncover adhoc tests you hadn’t thought of earlier.
• It ensures that internal modules and components communicate properly.
• It performs regression testing on important connection points.

How to write Integration Testing?

• Write a test plan.
• Create test cases and use cases.
• Run tests after unit integration.
• Detect errors.
• Retest the functionality after bug fixing.
• Repeat the testing cycle until all bugs are fixed.

Integration in the new Horizons Framework

• Follow The API first design approach to write your router, schema first
• Use the data model defined in schemas.py
• Use a test database with seed data and it must be version controlled

Integration Test Example

Test add Item API endpoint, what we excepted:

• Use the request model to post data to the POST /api/v1/items/ endpoint
• Use the response model to serialize the HTTP Response
• The return code must be success
• The output data must exactly match the input data
• Can get the data from the Get /api/v1/items/:id endpoint

Integration Test Example

Define our api and data model first

API endpoint app/item/router.py

@router.post("/", response_model=Response[ItemResponse])
async def create_item(
    request: ItemRequest, user_id: int = Depends(get_user_id)
):
    pass


@router.get("/{item_id}", response_model=Response[ItemResponse])
async def get_item(item_id: int, user_id: int = Depends(get_user_id)):
    pass
@router.post("/", response_model=Response[ItemResponse])
async def create_item(
    request: ItemRequest, user_id: int = Depends(get_user_id)
):
    pass


@router.get("/{item_id}", response_model=Response[ItemResponse])
async def get_item(item_id: int, user_id: int = Depends(get_user_id)):
    pass

Integration Test Example

Define our api and data model first

Data model app/item/schemas.py

from .models import Item
from horizons.schema_plus.models import BaseModel
from tortoise.contrib.pydantic import (
    pydantic_model_creator,
    pydantic_queryset_creator,
)


class ItemRequest(BaseModel):
    name: str
    description: str


ItemResponse = pydantic_model_creator(Item)
ItemsResponse = pydantic_queryset_creator(Item)
from .models import Item
from horizons.schema_plus.models import BaseModel
from tortoise.contrib.pydantic import (
    pydantic_model_creator,
    pydantic_queryset_creator,
)


class ItemRequest(BaseModel):
    name: str
    description: str


ItemResponse = pydantic_model_creator(Item)
ItemsResponse = pydantic_queryset_creator(Item)

Integration Test Example

check your api

Integration Test Example

Write the test case first in tests/integration/item/test_main.py

@pytest.mark.anyio
async def test_add_and_get_item(client_with_token):
    name = "item1"
    description = "description1"
    # use the request data model to post data
    item = ItemRequest(name=name, description=description)
    response = await client_with_token.post("/api/v1/items/", json=item.dict())
    assert response.status_code == 200
    # use the response data model to serialize data
    response = Response[ItemResponse].parse_raw(response.content)

    # the return code must be success
    assert response
    assert response.code == BusinessCode.NO_ERROR
    assert response.message == HTTPMessage.SUCCESS.value

    # The output data must exactly match the input data
    id = response.data.id
    assert id
    assert response.data.name == name
    assert response.data.description == description

    ...
@pytest.mark.anyio
async def test_add_and_get_item(client_with_token):
    name = "item1"
    description = "description1"
    # use the request data model to post data
    item = ItemRequest(name=name, description=description)
    response = await client_with_token.post("/api/v1/items/", json=item.dict())
    assert response.status_code == 200
    # use the response data model to serialize data
    response = Response[ItemResponse].parse_raw(response.content)

    # the return code must be success
    assert response
    assert response.code == BusinessCode.NO_ERROR
    assert response.message == HTTPMessage.SUCCESS.value

    # The output data must exactly match the input data
    id = response.data.id
    assert id
    assert response.data.name == name
    assert response.data.description == description

    ...

Integration Test Example

Write the test case first in tests/integration/item/test_main.py

@pytest.mark.anyio
async def test_add_and_get_item(client_with_token):
    name = "item1"
    description = "description1"
    
    ...

    response = await client_with_token.get(f"/api/v1/items/{id}")
    assert response.status_code == 200

    # use the response data model to serialize data
    response = Response[ItemResponse].parse_raw(response.content)

    # the return code must be success
    assert response
    assert response.code == BusinessCode.NO_ERROR
    assert response.message == HTTPMessage.SUCCESS.value

    # The output data must exactly match the input data
    assert id == response.data.id
    assert response.data.name == name
    assert response.data.description == description
@pytest.mark.anyio
async def test_add_and_get_item(client_with_token):
    name = "item1"
    description = "description1"
    
    ...

    response = await client_with_token.get(f"/api/v1/items/{id}")
    assert response.status_code == 200

    # use the response data model to serialize data
    response = Response[ItemResponse].parse_raw(response.content)

    # the return code must be success
    assert response
    assert response.code == BusinessCode.NO_ERROR
    assert response.message == HTTPMessage.SUCCESS.value

    # The output data must exactly match the input data
    assert id == response.data.id
    assert response.data.name == name
    assert response.data.description == description

Integration Test Example

./run_test.sh
./run_test.sh

the script will create a test database with seed data

Integration Test Example

write the service logic

class ItemService:
    async def add_item(
        self, user_id: int, request: ItemRequest
    ) -> ItemResponse:
        item = Item(**request.dict())
        item.user_id = user_id
        await Item.save(item)
        return await ItemResponse.from_tortoise_orm(item)

    async def get_item(self, user_id: int, item_id: int) -> ItemResponse:
        queryset = Item.filter(user_id=user_id, id=item_id).first()
        return await ItemResponse.from_queryset_single(queryset)
class ItemService:
    async def add_item(
        self, user_id: int, request: ItemRequest
    ) -> ItemResponse:
        item = Item(**request.dict())
        item.user_id = user_id
        await Item.save(item)
        return await ItemResponse.from_tortoise_orm(item)

    async def get_item(self, user_id: int, item_id: int) -> ItemResponse:
        queryset = Item.filter(user_id=user_id, id=item_id).first()
        return await ItemResponse.from_queryset_single(queryset)

Integration Test Example

update router.py and test again

@router.post("/", response_model=Response[ItemResponse])
async def create_item(
    request: ItemRequest, user_id: int = Depends(get_user_id)
):
    data = await item_service.add_item(user_id, request)
    return json_response(data=data)


@router.get("/{item_id}", response_model=Response[ItemResponse])
async def get_item(item_id: int, user_id: int = Depends(get_user_id)):
    data = await item_service.get_item(user_id, item_id)
    return json_response(data=data)
@router.post("/", response_model=Response[ItemResponse])
async def create_item(
    request: ItemRequest, user_id: int = Depends(get_user_id)
):
    data = await item_service.add_item(user_id, request)
    return json_response(data=data)


@router.get("/{item_id}", response_model=Response[ItemResponse])
async def get_item(item_id: int, user_id: int = Depends(get_user_id)):
    data = await item_service.get_item(user_id, item_id)
    return json_response(data=data)

Integration Test Example

Run the test again

./run_test.sh
./run_test.sh

we are great to see it’s success

Try it by yourself

• unit test code can be found at https://gitlab.com/rocketg/service/sample/-/tree/sample_unit_test
• integration test code can be found at https://gitlab.com/rocketg/service/sample/-/tree/integration_test

Exercises

Fire a pull request to the sample project with unit test and integration test codes

• write a unit test for any case
• add extra test for the integration tesst test_add_and_get_item
  • test the added data can be get from the list enndoint Get /api/v1/items
  • test the data can be delete from the Delete api/va/items/{item_id}
  • test again to make sure the data can not be get fron the list endpoint