與真實外部系統 Postgres 的整合測試
在軟體開發中，整合測試扮演著至關重要的角色，它驗證了不同模組或系統之間的互動是否符合預期。本篇文章將探討如何使用 pytest 框架和 Docker 容器化技術，對與 Postgres 資料函式庫整合的系統進行有效的整合測試。
我將分享我在實際專案中整合 Postgres 的經驗，並提供一些我在測試過程中學到的技巧和最佳實踐。
使用 pytest 標記整合測試
為了有效管理測試套件，我們可以使用 pytest 的標記功能來區分整合測試和其他型別的測試。
# tests/repository/postgres/test_postgresrepo.py
import pytest
pytestmark = pytest.mark.integration
def test_database_connection():
# 測試資料函式庫連線
pass
在這個例子中，pytestmark = pytest.mark.integration 將模組中的所有測試都標記為 integration。我們還需要在 pytest.ini 檔案中註冊這個標記：
# pytest.ini
[pytest]
markers =
integration: integration tests
現在，您可以使用 pytest -svv -m integration 命令來單獨執行標記為 integration 的測試。
跳過預設的整合測試
為了避免在日常開發中頻繁執行耗時的整合測試，我們可以設定 pytest 在預設情況下跳過這些測試，並提供一個命令列選項來手動啟用它們。
# tests/conftest.py
def pytest_addoption(parser):
parser.addoption(
"--integration", action="store_true", help="run integration tests"
)
def pytest_runtest_setup(item):
if "integration" in item.keywords and not item.config.getvalue("integration"):
pytest.skip("需要使用 --integration 選項來執行")
這段程式碼新增了一個 --integration 命令列選項。當指定此選項時，pytest 將執行標記為 integration 的測試；否則，這些測試將被跳過。
使用 Docker 協調 Postgres 容器
在執行整合測試時，我們需要一個正在執行的 Postgres 資料函式庫。Docker 提供了一個便捷的解決方案，可以輕鬆地建立和管理 Postgres 容器。
以下是一個使用 Docker Compose 啟動 Postgres 容器的範例 docker-compose.yml 檔案：
version: '3.8'
services:
postgres:
image: postgres:latest
ports:
- "5432:5432"
environment:
POSTGRES_USER: user
POSTGRES_PASSWORD: password
POSTGRES_DB: test_db
您可以使用 docker-compose up -d 命令啟動容器，並使用 docker-compose down 命令停止容器。
使用 SQLAlchemy 建立資料函式庫模型
SQLAlchemy 是一個功能強大的 Python ORM 框架，可以簡化資料函式庫操作。我們可以使用 SQLAlchemy 來定義資料函式庫表格的結構。
# rentomatic/repository/postgres_objects.py
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Room(Base):
__tablename__ = 'room'
id = Column(Integer, primary_key=True)
code = Column(String(36), nullable=False)
size = Column(Integer)
price = Column(Integer)
longitude = Column(Float)
latitude = Column(Float)
這個程式碼定義了一個 Room 類別，它對應資料函式庫中的 room 表格。需要注意的是，這個類別的結構是由儲存層的需求決定的，而不是由業務邏輯決定的。
graph LR
B[B]
E[E]
G[G]
A[pytest 標記] --> B{pytest.ini 註冊標記}
B --> C[執行整合測試]
D[Docker Compose] --> E{啟動 Postgres 容器}
E --> C
F[SQLAlchemy] --> G{定義資料函式庫模型}
G --> C
內容解密： 上面的 Mermaid 圖表展示了 pytest、Docker Compose 和 SQLAlchemy 如何協同工作以實作 Postgres 整合測試。首先，使用 pytest 標記整合測試，並在 pytest.ini 中註冊標記。然後，使用 Docker Compose 啟動 Postgres 容器。最後，使用 SQLAlchemy 定義資料函式庫模型。所有這些步驟都是執行整合測試的必要條件。
總結：本篇文章介紹瞭如何使用 pytest、Docker 和 SQLAlchemy 進行 Postgres 整合測試。透過結合這些工具和技術，您可以構建更健壯、更可靠的軟體系統。
在現代軟體開發中，容器化技術與資料函式庫的整合至關重要。本文將引導您使用 Docker Compose 建立 PostgreSQL 資料函式庫，並且 Python 應用程式無縫整合，同時示範如何利用 Click 框架簡化管理指令碼，並結合 Pytest 進行自動化測試。
首先，將 docker-compose 加入 requirements/test.txt：
-r prod.txt tox coverage pytest pytest-cov pytest-flask docker-compose
接著，執行 pip install -r requirements/dev.txt 安裝必要套件。以下為管理指令碼 manage.py 的完整程式碼：
#! /usr/bin/env python
import os
import json
import subprocess
import time
import click
import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT
# 設定環境變數
def setenv(variable, default):
os.environ[variable] = os.getenv(variable, default)
APPLICATION_CONFIG_PATH = "config"
DOCKER_PATH = "docker"
def app_config_file(config):
return os.path.join(APPLICATION_CONFIG_PATH, f"{config}.json")
def docker_compose_file(config):
return os.path.join(DOCKER_PATH, f"{config}.yml")
def read_json_configuration(config):
with open(app_config_file(config)) as f:
config_data = json.load(f)
return dict((i["name"], i["value"]) for i in config_data)
def configure_app(config):
configuration = read_json_configuration(config)
for key, value in configuration.items():
setenv(key, value)
@click.group()
def cli():
pass
def docker_compose_cmdline(commands_string=None):
config = os.getenv("APPLICATION_CONFIG")
configure_app(config)
compose_file = docker_compose_file(config)
if not os.path.isfile(compose_file):
raise ValueError(f"檔案 {compose_file} 不存在")
command_line = [
"docker-compose",
"-p",
config,
"-f",
compose_file,
]
if commands_string:
command_line.extend(commands_string.split(" "))
return command_line
def run_sql(statements):
conn = psycopg2.connect(
dbname=os.getenv("POSTGRES_DB"),
user=os.getenv("POSTGRES_USER"),
password=os.getenv("POSTGRES_PASSWORD"),
host=os.getenv("POSTGRES_HOSTNAME"),
port=os.getenv("POSTGRES_PORT"),
)
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cursor = conn.cursor()
for statement in statements:
cursor.execute(statement)
cursor.close()
conn.close()
def wait_for_logs(cmdline, message):
logs = subprocess.check_output(cmdline)
while message not in logs.decode("utf-8"):
time.sleep(1)
logs = subprocess.check_output(cmdline)
@cli.command()
@click.argument("args", nargs=-1)
def test(args):
os.environ["APPLICATION_CONFIG"] = "testing"
configure_app(os.getenv("APPLICATION_CONFIG"))
cmdline = docker_compose_cmdline("up -d")
subprocess.call(cmdline)
cmdline = docker_compose_cmdline("logs postgres")
wait_for_logs(cmdline, "ready to accept connections")
run_sql([f"CREATE DATABASE {os.getenv('APPLICATION_DB')}"])
cmdline = [
"pytest",
"-svv",
"--cov=application",
"--cov-report=term-missing",
]
cmdline.extend(args)
subprocess.call(cmdline)
cmdline = docker_compose_cmdline("down")
subprocess.call(cmdline)
if __name__ == "__main__":
cli()
內容解密：
此指令碼利用 Click 建立命令列介面，方便管理 Docker Compose 與 PostgreSQL。docker_compose_cmdline 函式簡化了 Docker Compose 命令的組裝，run_sql 函式則用於執行 SQL 指令，wait_for_logs 函式確保 PostgreSQL 容器啟動完成。test 命令則整合了測試流程，先啟動容器、建立測試資料函式庫，再執行 Pytest。
graph LR
B[B]
A[設定環境變數] --> B{讀取JSON設定};
B --> C[設定應用程式];
C --> D[執行Docker Compose];
D --> E[等待PostgreSQL啟動];
E --> F[建立測試資料函式庫];
F --> G[執行Pytest];
G --> H[停止Docker Compose];
圖表說明： 此流程圖展示了測試流程的執行步驟，從設定環境變數開始，到最後停止 Docker Compose。
classDiagram
class ManageScript {
+setenv(variable, default)
+app_config_file(config)
+docker_compose_file(config)
+read_json_configuration(config)
+configure_app(config)
+docker_compose_cmdline(commands_string)
+run_sql(statements)
+wait_for_logs(cmdline, message)
+test(args)
}
圖表說明： 此類別圖展示了 manage.py 指令碼中定義的類別與方法，清晰地呈現了指令碼的結構。
透過以上步驟，我們成功地整合了 Docker Compose 和 PostgreSQL，並利用 Python 建立了簡潔易用的管理指令碼。此架構不僅提升了開發效率，也確保了測試環境的一致性，為開發更穩健的應用程式奠定了基礎。
在實際應用中，可以根據專案需求調整設定檔和 SQL 指令。此架構也適用於其他資料函式庫，只需修改連線設定和 SQL 語法即可。
透過容器化技術和自動化測試，我們可以更有效率地開發和佈署應用程式，並確保程式碼品質和系統穩定性。
與真實外部系統整合：PostgreSQL實戰
在應用程式開發過程中，與外部系統的整合測試至關重要。本文將以PostgreSQL為例，分享我在實際專案中整合測試的經驗與技巧。
應用程式設定名稱為testing，表示使用config/testing.json設定檔和docker/testing.yml Docker Compose檔案。這些命名和路徑只是管理指令碼的慣例設定，您可以根據專案需求調整。
函式根據Docker Compose檔案啟動容器，執行docker-compose up -d。等待資料函式庫準備就緒的日誌訊息後，執行建立測試資料函式庫的SQL命令。接著，使用預設選項執行Pytest，加入所有命令列提供的選項，最後關閉Docker Compose容器。
graph LR
B[B]
No[No]
Yes[Yes]
A[啟動 Docker 容器] --> B{資料函式庫準備就緒?};
B -- Yes --> C[執行 SQL 建立測試資料函式庫];
B -- No --> B;
C --> D[執行 Pytest];
D --> E[關閉 Docker 容器];
設定檔詳解
為了完成設定，需要定義Docker Compose設定檔docker/testing.yml：
version: '3.8'
services:
postgres:
image: postgres
environment:
POSTGRES_DB: ${POSTGRES_DB}
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
ports:
- "${POSTGRES_PORT}:5432"
以及JSON設定檔config/testing.json：
[
{
"name": "FLASK_ENV",
"value": "production"
},
{
"name": "FLASK_CONFIG",
"value": "testing"
},
{
"name": "POSTGRES_DB",
"value": "postgres"
},
{
"name": "POSTGRES_USER",
"value": "postgres"
},
{
"name": "POSTGRES_HOSTNAME",
"value": "localhost"
},
{
"name": "POSTGRES_PORT",
"value": "5433"
},
{
"name": "POSTGRES_PASSWORD",
"value": "postgres"
},
{
"name": "APPLICATION_DB",
"value": "test"
}
]
這個設定定義了FLASK_ENV和FLASK_CONFIG。FLASK_ENV是Flask內部變數，只能是development或production，與內部除錯器相關。FLASK_CONFIG用於設定Flask應用程式。測試時，將FLASK_ENV設為production，FLASK_CONFIG設為testing。
其他JSON設定初始化以POSTGRES_開頭的變數，這些是PostgreSQL Docker容器所需的變數。容器執行時，會自動建立名稱由POSTGRES_DB指定的資料函式庫，並使用POSTGRES_USER和POSTGRES_PASSWORD的值建立使用者和密碼。
我引入了APPLICATION_DB變數，因為我想建立一個不同於預設資料函式庫的特定資料函式庫。預設埠POSTGRES_PORT從標準值5432更改為5433，避免與機器上執行的其他資料函式庫衝突。Docker Compose設定檔中，這只更改容器的外部對映，不影響資料函式庫引擎在容器內使用的實際埠。
資料函式庫Fixture設定
由於在JSON檔案中定義了資料函式庫設定，需要一個fixture來載入相同的設定，以便在測試期間連線到資料函式庫。
# tests/conftest.py
from manage import read_json_configuration
@pytest.fixture(scope="session")
def app_configuration():
return read_json_configuration("testing")
為了簡潔起見，我硬編碼了設定檔的名稱。另一種解決方案是在管理指令碼中建立一個包含應用程式設定的環境變數，並從這裡讀取。
其餘的fixture包含特定於PostgreSQL的程式碼，因此最好將程式碼放在更具體的檔案conftest.py中：
# tests/repository/postgres/conftest.py
import sqlalchemy
import pytest
from rentomatic.repository.postgres_objects import Base, Room
@pytest.fixture(scope="session")
def pg_session_empty(app_configuration):
# ... (連線字串與資料函式庫引擎設定)
DBSession = sqlalchemy.orm.sessionmaker(bind=engine)
session = DBSession()
yield session
session.close()
connection.close
@pytest.fixture(scope="session")
def pg_test_data():
return [
# ... (測試資料)
]
@pytest.fixture(scope="function")
def pg_session(pg_session_empty, pg_test_data):
for r in pg_test_data:
new_room = Room(
# ... (房間資料)
)
pg_session_empty.add(new_room)
pg_session_empty.commit()
yield pg_session_empty
pg_session_empty.query(Room).delete()
第一個fixture pg_session_empty建立一個到空初始資料函式庫的session，而pg_test_data定義了將載入到資料函式庫中的值。由於我們沒有改變這組值，因此不需要建立fixture，但這是讓其他fixture和測試都能使用它的最簡單方法。最後一個fixture pg_session使用測試資料建立的PostgreSQL物件填充資料函式庫。請注意，這些不是實體，而是我們建立的用於對映它們的PostgreSQL物件。
請注意，最後一個fixture的作用域是函式級別的，因此它會為每個測試執行。因此，我們在yield傳回後刪除所有房間，使資料函式庫還原到測試前的狀態。一般來說，您應該始終在測試後清理。我們正在測試的端點不會寫入資料函式庫，所以在這種特定情況下，並不需要清理，但我更喜歡從零開始實作一個完整的解決方案。
我們可以透過更改函式test_dummy來測試整個設定，使其擷取表格Room的所有列，並驗證查詢是否傳回4個值。
tests/repository/postgres/test_postgresrepo.py的新版本是：
import pytest
from rentomatic.repository.postgres_objects import Room
pytestmark = pytest.mark.integration
def test_dummy(pg_session):
assert len(pg_session.query(Room).all()) == 4
此時，您可以使用整合測試執行測試套件。當pytest執行函式test_dummy時，您應該會注意到明顯的延遲，因為Docker需要一些時間來啟動資料函式庫容器並準備資料。
內容解密：
這段程式碼展示瞭如何使用 pytest 進行整合測試，特別是如何與 PostgreSQL 資料函式庫互動。pg_session fixture 負責建立資料函式庫連線、新增測試資料，並在測試結束後清理資料函式庫。test_dummy 函式則驗證資料函式庫中房間的數量是否正確。透過這種方式，我們可以確保程式碼與真實的外部系統正確整合。
與真實外部系統整合：MongoDB
在軟體開發中，整合外部系統是不可避免的。前一章我們探討瞭如何將真實的外部系統（PostgreSQL）整合到乾淨架構的核心。本章將延續前章的基礎，展示如何以 MongoDB 取代 PostgreSQL，體現乾淨架構的彈性，以及如何在同一架構下輕鬆整合不同的資料函式庫方法。
Fixtures 的設定與運用
乾淨架構的彈性讓支援多種儲存系統變得輕而易舉。在本章節中，我將實作 MongoRepo 類別，提供與知名 NoSQL 資料函式庫 MongoDB 的介面。我們將沿用與 PostgreSQL 相同的測試策略，使用 Docker 容器執行資料函式庫，並使用 docker-compose 進行協調。
以下程式碼片段展示瞭如何設定 pytest fixtures，用於建立和銷毀測試用的 MongoDB 資料函式庫：
import pytest
import pymongo
@pytest.fixture(scope="session")
def mongo_client():
# 連線到 MongoDB 測試容器
client = pymongo.MongoClient("mongodb://test_user:test_password@localhost:27017/")
yield client
client.close()
@pytest.fixture(scope="function")
def mongo_db(mongo_client):
# 建立測試資料函式庫
db = mongo_client["test_db"]
yield db
db.drop_collection("rooms") # 清除測試資料
@pytest.fixture(scope="function")
def mongo_test_data(mongo_db):
# 插入測試資料
rooms_data = [
{"code": "room001", "size": 20, "price": 100, "latitude": 34.0522, "longitude": -118.2437},
{"code": "room002", "size": 30, "price": 150, "latitude": 37.7749, "longitude": -122.4194},
# ... 其他測試資料
]
mongo_db["rooms"].insert_many(rooms_data)
return rooms_data
內容解密：
這段程式碼定義了三個 pytest fixtures：mongo_client、mongo_db 和 mongo_test_data。mongo_client 建立一個連線到 MongoDB 測試容器的 client，並在測試結束後關閉連線。mongo_db 建立一個名為 test_db 的測試資料函式庫，並在每個測試函式執行後清除其中的 rooms 集合。mongo_test_data 則插入測試資料到 rooms 集合中，並將資料傳回給測試函式使用。這些 fixtures 確保了每個測試都在乾淨的環境中執行，避免了測試資料之間的相互幹擾。
MongoRepo 的實作
MongoRepo 類別的實作與 PostgresRepo 類別似，主要區別在於資料函式庫操作的語法。以下程式碼片段展示了 MongoRepo 的 list 方法：
from rentomatic.domain import room
class MongoRepo:
def __init__(self, configuration):
self.client = pymongo.MongoClient(configuration["MONGODB_URI"])
self.db = self.client[configuration["MONGODB_DB"]]
self.collection = self.db["rooms"]
def _create_room_objects(self, results):
return [
room.Room(
code=q["code"],
size=q["size"],
price=q["price"],
latitude=q["latitude"],
longitude=q["longitude"],
)
for q in results
]
def list(self, filters=None):
query = {}
if filters:
if "code__eq" in filters:
query["code"] = filters["code__eq"]
if "price__eq" in filters:
query["price"] = filters["price__eq"]
# ... 其他篩選條件
results = self.collection.find(query)
return self._create_room_objects(results)
內容解密：
MongoRepo 的 __init__ 方法建立 MongoDB 連線，並指定要使用的資料函式庫和集合。_create_room_objects 方法將 MongoDB 的查詢結果轉換為 room.Room 物件。list 方法則根據提供的篩選條件查詢資料函式庫，並傳回 room.Room 物件的列表。
透過以上實作，我們可以無縫地將 MongoDB 整合到現有的乾淨架構中，展現了乾淨架構的彈性與可擴充套件性。
流程圖
以下流程圖展示了使用 MongoRepo 查詢房間資料的流程：
graph LR
A[開始] --> B{設定篩選條件};
B -- 有篩選條件 --> C[建立 MongoDB 查詢];
B -- 無篩選條件 --> D[查詢所有房間];
C --> E[執行 MongoDB 查詢];
D --> E;
E --> F[將結果轉換為 Room 物件];
F --> G[傳回 Room 物件列表];
G --> H[結束];
總結：本章展示瞭如何將 MongoDB 整合到乾淨架構中，並說明瞭如何使用 pytest fixtures 進行測試。透過這個案例，我們可以看到乾淨架構的彈性，以及如何輕鬆地支援不同的資料函式庫系統。透過使用 Mermaid 圖表，我們可以更清晰地理解程式碼的流程和邏輯。
在先前的文章中，我分享瞭如何建構複雜的測試結構。現在，我將運用這個架構，為新的儲存系統（MongoDB）實作測試。這套測試結構的優勢在於可以重複使用既有的 pytest fixtures，大幅提升開發效率。
## 建立 MongoDB 的 pytest fixtures
首先，我們定義 `tests/repository/mongodb/conftest.py` 檔案，其中包含 MongoDB 的 pytest fixtures，與先前為 PostgreSQL 建立的檔案結構相似。
```python
import pymongo
import pytest
@pytest.fixture(scope="session")
def mg_database_empty(app_configuration):
client = pymongo.MongoClient(
host=app_configuration["MONGODB_HOSTNAME"],
port=int(app_configuration["MONGODB_PORT"]),
username=app_configuration["MONGODB_USER"],
password=app_configuration["MONGODB_PASSWORD"],
authSource="admin",
)
db = client[app_configuration["APPLICATION_DB"]]
yield db
client.drop_database(app_configuration["APPLICATION_DB"])
client.close()
@pytest.fixture(scope="function")
def mg_test_data():
return [
{"code": "f853578c-fc0f-4e65-81b8-566c5dffa35a", "size": 215, "price": 39, "longitude": -0.09998975, "latitude": 51.75436293},
{"code": "fe2c3195-aeff-487a-a08f-e0bdc0ec6e9a", "size": 405, "price": 66, "longitude": 0.18228006, "latitude": 51.74640997},
{"code": "913694c6-435a-4366-ba0d-da5334a611b2", "size": 56, "price": 60, "longitude": 0.27891577, "latitude": 51.45994069},
{"code": "eed76e77-55c1-41ce-985d-ca49bf6c0585", "size": 93, "price": 48, "longitude": 0.33894476, "latitude": 51.39916678},
]
@pytest.fixture(scope="function")
def mg_database(mg_database_empty, mg_test_data):
collection = mg_database_empty.rooms
collection.insert_many(mg_test_data)
yield mg_database_empty
collection.delete_many({})
內容解密：
這些函式與先前為 PostgreSQL 定義的函式非常相似。mg_database_empty 負責建立 MongoDB 使用者端和空的資料函式庫，並在 yield 後釋放資源。mg_test_data 提供測試資料，而 mg_database 將測試資料填入空的資料函式庫。雖然 SQLAlchemy 套件透過 session 運作，而 PyMongo 函式庫直接建立和使用 client，但整體結構是相同的。
更新專案需求
由於我們使用了 PyMongo 函式庫，因此需要更新生產環境的需求檔案 requirements/prod.txt：
Flask SQLAlchemy psycopg2 pymongo
然後執行 pip install -r requirements/dev.txt 安裝新的依賴套件。
Docker Compose 設定
我們需要在測試用的 Docker Compose 設定檔中新增一個 MongoDB 容器。MongoDB image 只需要 MONGO_INITDB_ROOT_USERNAME 和 MONGO_INITDB_ROOT_PASSWORD 這兩個環境變數，因為它不會建立任何初始資料函式庫。如同 PostgreSQL 容器的設定，我們也為 MongoDB 容器分配一個非標準的連線埠，以便在其他容器執行時也能執行測試。
version: '3.8'
services:
postgres:
image: postgres
environment:
POSTGRES_DB: ${POSTGRES_DB}
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
ports:
- "${POSTGRES_PORT}:5432"
mongo:
image: mongo
environment:
MONGO_INITDB_ROOT_USERNAME: ${MONGODB_USER}
MONGO_INITDB_ROOT_PASSWORD: ${MONGODB_PASSWORD}
ports:
- "${MONGODB_PORT}:27017"
內容解密：
這個 Docker Compose 設定檔定義了兩個服務：postgres 和 mongo。postgres 服務使用 PostgreSQL image，並設定了必要的環境變數，例如資料函式庫名稱、使用者名稱和密碼。同時，它將容器的 5432 連線埠對映到主機的 ${POSTGRES_PORT}。mongo 服務使用 MongoDB image，並設定了 root 使用者名稱和密碼。它將容器的 27017 連線埠對映到主機的 ${MONGODB_PORT}。
應用程式設定
Docker Compose、測試框架和應用程式本身都透過單個 JSON 檔案進行設定，我們需要使用 MongoDB 的實際值來更新它。
[
{"name": "FLASK_ENV", "value": "production"},
{"name": "FLASK_CONFIG", "value": "testing"},
{"name": "POSTGRES_DB", "value": "postgres"},
{"name": "POSTGRES_USER", "value": "postgres"},
{"name": "POSTGRES_HOSTNAME", "value": "localhost"},
{"name": "POSTGRES_PORT", "value": "5433"},
{"name": "POSTGRES_PASSWORD", "value": "postgres"},
{"name": "MONGODB_USER", "value": "root"},
{"name": "MONGODB_HOSTNAME", "value": "localhost"},
{"name": "MONGODB_PORT", "value": "27018"},
{"name": "MONGODB_PASSWORD", "value": "mongodb"},
{"name": "APPLICATION_DB", "value": "test"}
]
內容解密：
此設定檔使用 JSON 格式定義了一系列環境變數。每個變數都包含 name 和 value 兩個欄位。這些變數涵蓋了 Flask 應用程式、PostgreSQL 資料函式庫和 MongoDB 資料函式庫的設定。例如，FLASK_ENV 設定為 production，POSTGRES_PORT 設定為 5433，MONGODB_PORT 設定為 27018，APPLICATION_DB 設定為 test。
由於 MongoDB 的標準連線埠是 27017，我選擇使用 27018 作為測試連線埠。請記住，這只是一個範例。在實際情況中，我們可能有不同的環境和不同的測試設定，此時我們可能需要為容器分配一個隨機連線埠，並使用 Python 擷取該值並將其傳遞給應用程式。
另外，我選擇使用相同的變數 APPLICATION_DB 作為 PostgreSQL 和 MongoDB 資料函式庫的名稱。同樣，這只是一個簡單的範例，在更複雜的情況下，您的做法可能有所不同。
整合測試的設計與實踐
接下來，我們將探討整合測試的設計與實踐。整合測試的目標是驗證不同系統元件之間的互動是否正常運作。在這個案例中，我們將測試應用程式與 MongoDB 的整合。
我編寫的整合測試與 PostgreSQL 的測試非常相似，因為它們涵蓋了相同的使用案例。如果您在同一個系統中使用多個資料函式庫，您可能需要服務不同的使用案例，因此在實際案例中，這一步驟可能會更複雜。然而，如果您只是想提供對多個資料函式庫的支援，讓您的客戶端可以選擇插入到系統中，那麼您完全可以像我一樣，複製並調整相同的測試套件。
import pytest
from rentomatic.repository import mongorepo
pytestmark = pytest.mark.integration
def test_repository_list_without_parameters(app_configuration, mg_database, mg_test_data):
repo = mongorepo.MongoRepo(app_configuration)
repo_rooms = repo.list()
assert set([r.code for r in repo_rooms]) == set([r["code"] for r in mg_test_data])
# ... 其他測試函式 ...
內容解密：
這段程式碼展示了一個名為 test_repository_list_without_parameters 的整合測試函式。它使用了 pytest 框架，並標記為 integration。測試函式接受 app_configuration、mg_database 和 mg_test_data 三個 fixture 作為引數。它首先建立一個 MongoRepo 物件，然後呼叫 list() 方法取得所有房間資料。最後，它斷言從資料函式庫取得的房間程式碼集合與測試資料中的房間程式碼集合相同。
我新增了一個名為 test_repository_list_with_price_as_string 的測試，用於檢查當篩選條件中的價格以字串表示時會發生什麼。透過使用 MongoDB shell 進行實驗，我發現這種情況下查詢無法正常運作，因此我加入了這個測試，以確保實作不會出現問題。
透過以上步驟，我們成功地將 MongoDB 整合到系統中，並建立了完善的測試機制。這確保了系統的穩定性和可靠性，同時也提升了開發效率。
import pymongo
from rentomatic.domain import room
class MongoRepo:
def __init__(self, configuration):
client = pymongo.MongoClient(
host=configuration["MONGODB_HOSTNAME"],
port=int(configuration["MONGODB_PORT"]),
username=configuration["MONGODB_USER"],
password=configuration["MONGODB_PASSWORD"],
authSource="admin",
)
self.db = client[configuration["APPLICATION_DB"]]
def _create_room_objects(self, results):
return [
room.Room(
code=q["code"],
size=q["size"],
price=q["price"],
latitude=q["latitude"],
longitude=q["longitude"],
)
for q in results
]
def list(self, filters=None):
collection = self.db.rooms
if filters is None:
result = collection.find()
else:
mongo_filter = {}
for key, value in filters.items():
key, operator = key.split("__")
filter_value = mongo_filter.get(key, {})
if key == "price":
value = int(value)
filter_value["${}".format(operator)] = value
mongo_filter[key] = filter_value
result = collection.find(mongo_filter)
return self._create_room_objects(result)
內容解密：
這段程式碼定義了一個 MongoRepo 類別，用於與 MongoDB 進行互動，實作了儲存函式庫（Repository）模式。它接收一個組態字典作為引數，用於建立 MongoDB 連線。_create_room_objects 方法將 MongoDB 查詢結果轉換為 Room 物件列表。list 方法則根據提供的篩選條件查詢房間資料，並傳回 Room 物件列表。篩選條件的設計巧妙地利用了 MongoDB 的查詢運算元，例如 price__gt 可以轉換為 MongoDB 的 $gt 運算元。值得注意的是，程式碼中對價格進行了型別轉換，確保與 MongoDB 中的資料型別一致。這種設計使得儲存函式庫層能夠靈活地適應不同的資料函式庫後端，而上層應用程式碼無需關心底層資料函式庫的具體實作。
graph LR
C[C]
A[應用程式] --> B(儲存函式庫層)
B --> C{MongoDB}
D[篩選條件] --> B
B --> E[Room物件列表]
圖表說明: 此流程圖展示了應用程式、儲存函式庫層和 MongoDB 之間的互動關係。應用程式透過儲存函式庫層存取 MongoDB，並根據篩選條件取得 Room 物件列表。
[
{
"name": "FLASK_ENV",
"value": "production"
},
{
"name": "FLASK_CONFIG",
"value": "production"
},
{
"name": "POSTGRES_DB",
"value": "postgres"
},
{
"name": "POSTGRES_USER",
"value": "postgres"
},
{
"name": "POSTGRES_HOSTNAME",
"value": "localhost"
},
{
"name": "POSTGRES_PORT",
"value": "5432"
},
{
"name": "POSTGRES_PASSWORD",
"value": "postgres"
},
{
"name": "APPLICATION_DB",
"value": "application"
}
]
內容解密：
這段 JSON 設定了應用程式在生產環境中的環境變數。FLASK_ENV 和 FLASK_CONFIG 設定為 production，表示應用程式將以生產模式執行。其他變數則設定了 PostgreSQL 資料函式庫的連線資訊，包括資料函式庫名稱、使用者名稱、主機名稱、連線埠和密碼。APPLICATION_DB 設定為 application，指定了應用程式使用的資料函式庫名稱。
version: '3.8'
services:
db:
image: postgres
environment:
POSTGRES_DB: ${POSTGRES_DB}
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
ports:
- "${POSTGRES_PORT}:5432"
volumes:
- pgdata:/var/lib/postgresql/data
web:
build:
context: ${PWD}
dockerfile: docker/web/Dockerfile.production
environment:
FLASK_ENV: ${FLASK_ENV}
FLASK_CONFIG: ${FLASK_CONFIG}
APPLICATION_DB: ${APPLICATION_DB}
POSTGRES_USER: ${POSTGRES_USER}
POSTGRES_HOSTNAME: "db"
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_PORT: ${POSTGRES_PORT}
command: gunicorn -w 4 -b 0.0.0.0 wsgi:app
volumes:
- ${PWD}:/opt/code
nginx:
image: nginx
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
ports:
- 8080:8080
volumes:
pgdata:
內容解密：
這段 Docker Compose 設定定義了生產環境中使用的服務，包括 PostgreSQL 資料函式庫、Flask 應用程式和 Nginx Web 伺服器。db 服務使用 PostgreSQL 映象，並設定了資料函式庫連線資訊。web 服務使用自定義的 Dockerfile 構建，並設定了環境變數和啟動命令，使用 Gunicorn 執行程式，並將目前的目錄掛載到容器中。nginx 服務使用 Nginx 映象，並將本機的 Nginx 設定檔掛載到容器中，設定連線埠為8080。 volumes 區段定義了 PostgreSQL 資料的持久化儲存。
建構一個產品級應用程式，需要整合多個元件，例如資料函式庫、Web 伺服器和應用程式本身。透過分層架構和完善的測試機制，可以確保系統的穩定性和可維護性。使用 Docker Compose 可以簡化佈署流程，並確保不同環境的一致性。
graph LR
subgraph 網際網路
A[使用者] --> B(Nginx)
end
B --> C(Gunicorn)
C --> D[Flask應用程式]
D --> E[Postgres資料函式庫]
圖表說明: 此架構圖展示了生產環境中各個元件的佈署關係，包含使用者、Nginx、Gunicorn、Flask 應用程式和 PostgreSQL 資料函式庫。
透過以上程式碼、設定和圖表，我們可以清晰地瞭解如何建構一個產品級的應用程式，並透過容器化技術進行佈署。
在現今的軟體開發環境中，容器化技術已經成為不可或缺的一部分。Docker 的出現，讓開發者可以更輕鬆地建構、佈署和管理應用程式。本文將分享我如何利用 Docker 建構一個可立即投入生產的 Web 應用程式，並著重於實務操作和程式碼範例。
