1. 环境说明

操作系统：macOS 12.6
Docker：20.10.17
Docker Compose：v2.10.2

2. 项目结构


├── Dockerfile
├── app.py
└── integration-test
    ├── Dockerfile
    ├── docker-compose.yml
    ├── test.py
    ├── test.sh
    └── .env

Dockerfile：


FROM python:3.10.10-alpine3.17

RUN pip3 install tortoise-orm==0.19.3 && \
    pip3 install fastapi==0.90.1 && \
    pip3 install uvicorn==0.20.0 && \
    pip3 install pydantic==1.10.4 && \
    pip3 install asyncmy==0.2.5 && \
    pip3 install aiomysql==0.1.1 && \
    apk update && \
    apk add curl

WORKDIR /
COPY app.py /app.py

CMD ["sh", "-c", "python app.py"]

app.py：


import os

import pydantic
from fastapi import FastAPI
import uvicorn
from tortoise.contrib.fastapi import register_tortoise
import tortoise


class ItemRequestSchema(pydantic.BaseModel):
    item_id: int = pydantic.Field(gt=0)
    item_name: str = pydantic.Field(max_length=128)


class ItemModel(tortoise.models.Model):
    item_id = tortoise.fields.IntField(null=False, pk=True)
    item_name = tortoise.fields.CharField(max_length=128)

    class Meta:
        table = "item"


app = FastAPI()


@app.get("/health")
async def health():
    return {"success": True}


@app.put("/create")
async def put(item: ItemRequestSchema):
    await ItemModel.create(item_id=item.item_id, item_name=item.item_name)
    return {"success": True}


def main():
    register_tortoise(
        app,
        db_url=f"mysql://root:{os.getenv('MYSQL_ROOT_PASSWORD')}@mysql:3306/item",
        modules={"models": ["__main__"]},
        generate_schemas=False,
        add_exception_handlers=True
    )
    uvicorn.run(app, host="0.0.0.0", port=8080)


if __name__ == "__main__":
    main()

integration-test/Dockerfile：


FROM python:3.10.10-alpine3.17

RUN pip3 install requests==2.28.1 && \
    pip3 install PyMySQL==1.0.2

WORKDIR /
COPY test.py /test.py
CMD ["sh", "-c", "python3 test.py"]

integration-test/docker-compose.yml：


version: "3.5"

services:
  # 启动数据库
  mysql:
    container_name: myproject-mysql
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
    image: mysql:8
    networks:
      - myproject-network
    # 健康检查，保证 MySQL 处于就绪
    healthcheck:
      test: mysql -uroot -proot -e "SELECT 1;" || exit 1
      start_period: 5s
      interval: 5s
      timeout: 5s
      retries: 20
  # 启动初始化阶段
  init-stage:
    # 依赖 mysql 服务就绪
    depends_on:
      mysql:
        condition: service_healthy
    container_name: myproject-init-stage
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
    image: mysql:8
    networks:
      - myproject-network
    # 创建数据库和表
    command: >+
           mysql -hmysql -uroot -p${MYSQL_ROOT_PASSWORD}
            -e "create database if not exists item;
                use item;
                create table if not exists item(
                    item_id int primary key,
                    item_name varchar(128)
                );"
  # 启动待测试服务
  myservice:
    # 依赖初始化阶段成功完成
    depends_on:
      init-stage:
        condition: service_completed_successfully
    container_name: myproject-myservice
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
    image: myservice:${TAG}
    build:
      context: ..
    networks:
      - myproject-network
    healthcheck:
      test: curl -sSL -m 1 -o /dev/null http://localhost:8080/health
      start_period: 2s
      interval: 2s
      timeout: 2s
      retries: 30
  # 启动集成测试
  integration-tester:
    # 依赖 myservice 服务就绪
    depends_on:
      myservice:
        condition: service_healthy
    container_name: myproject-integration-tester
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
    image: integration-tester:${TAG}
    build:
      context: .
    networks:
      - myproject-network

networks:
  myproject-network:
    driver: bridge
    name: myproject-network

integration-test/test.py：


import os
import unittest

import pymysql
import requests


class TestMyService(unittest.TestCase):
    def setUp(self) -> None:
        self.db = pymysql.connect(
            host="mysql",
            port=3306,
            user="root",
            password=os.getenv("MYSQL_ROOT_PASSWORD"),
            database="item"
        )
        self.base_url = "http://myservice:8080"

    def tearDown(self) -> None:
        self.db.close()

    def testCreate(self):
        item = {
            "item_id": 100,
            "item_name": "item_100"
        }
        # 请求 API，插入一条数据
        resp = requests.put(f"{self.base_url}/create", json=item)
        self.assertTrue(200 <= resp.status_code < 300)

        # 从 MySQL 中读数据
        cursor = self.db.cursor()
        cursor.execute(
            """
            SELECT `item_id`, `item_name` FROM `item` WHERE `item_id` = %s
            """,
            args=[item["item_id"]]
        )
        result = cursor.fetchone()
        self.assertEqual(result, (item["item_id"], item["item_name"]))
        cursor.close()


if __name__ == "__main__":
    unittest.main()

integration-test/test.sh：


#!/bin/bash

source .env

# define some colors to use for output
RED='\033[0;31m'
GREEN='\033[0;32m'
NC='\033[0m'

# kill and remove any running containers
cleanup () {
  docker-compose -p $PROJECT_NAME down
  # remove images
  docker image rm --force integration-tester:${TAG} myservice:${TAG}
}

# catch unexpected failures, do cleanup and output an error message
trap 'cleanup ; printf "${RED}Tests Failed For Unexpected Reasons${NC}\n"' \
  HUP INT QUIT PIPE TERM

# build and run the composed services
docker-compose -p $PROJECT_NAME build && docker-compose -p $PROJECT_NAME up -d
if [ $? -ne 0 ] ; then
  printf "${RED}Docker Compose Failed${NC}\n"
  exit -1
fi

# wait for the test service to complete and grab the exit code
TEST_EXIT_CODE=`docker wait myproject-integration-tester`
# output the logs for the test (for clarity)
docker logs myproject-integration-tester
# inspect the output of the test and display respective message
if [ -z "${TEST_EXIT_CODE}" ] || [ "$TEST_EXIT_CODE" -ne 0 ] ; then
  printf "${RED}Tests Failed${NC} - Exit Code: $TEST_EXIT_CODE\n"
else
  printf "${GREEN}Tests Passed${NC}\n"
fi

# call the cleanup fuction
cleanup
# exit the script with the same code as the test service code
exit $TEST_EXIT_CODE

integration-test/.env：


TAG=0.0.1
MYSQL_ROOT_PASSWORD=root
PROJECT_NAME=test-ci

3. 执行


xxxxxxxxxx
$ cd integration-test
$ chmod u+x test.sh
$ ./test.sh
# 检查测试脚本的退出码
$ echo $?

1. 环境说明

2. 项目结构

3. 执行

Read Also