From ca5f5fb06d6e7cd2e1322b2bdf3e80f9c74f6045 Mon Sep 17 00:00:00 2001 From: tech Date: Tue, 21 Apr 2026 16:21:07 +0800 Subject: [PATCH] feat: add integration tests and seed script for Azure Storage emulator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - scripts/seed_test_data.py: seeds 45 orders (30 Recived + 15 Payed) across 5 users and 3 days into the emulator - tests/test_integration.py: full pipeline test (seed → scan → backup → delete) with table cleanup before/after - pytest marker "integration" to separate from unit tests Co-Authored-By: Claude Opus 4.7 --- pyproject.toml | 5 ++ scripts/seed_test_data.py | 115 ++++++++++++++++++++++++++ tests/test_integration.py | 170 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 290 insertions(+) create mode 100644 scripts/seed_test_data.py create mode 100644 tests/test_integration.py diff --git a/pyproject.toml b/pyproject.toml index 095d88e..359ae4e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,3 +21,8 @@ packages = ["src/order_bak"] dev = [ "pytest>=8.0", ] + +[tool.pytest.ini_options] +markers = [ + "integration: integration tests requiring Azure Storage emulator", +] diff --git a/scripts/seed_test_data.py b/scripts/seed_test_data.py new file mode 100644 index 0000000..f84ddbd --- /dev/null +++ b/scripts/seed_test_data.py @@ -0,0 +1,115 @@ +"""Seed test data into Azure Table emulator for integration testing.""" + +import sys +from datetime import datetime, timezone + +from azure.data.tables import TableServiceClient + +DEFAULT_CONNECTION_STRING = ( + "AccountName=devstoreaccount1;" + "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/" + "K1SZFPTOtr/KBHBeksoGMGw==;" + "DefaultEndpointsProtocol=http;" + "BlobEndpoint=http://192.168.9.101:10000/devstoreaccount1;" + "QueueEndpoint=http://192.168.9.101:10001/devstoreaccount1;" + "TableEndpoint=http://192.168.9.101:10002/devstoreaccount1;" +) + +TABLE_NAME = "order" + +# Test users +USERS = ["user001", "user002", "user003", "user004", "user005"] + +# 3 days of data: Jan 1-3, 2025 +DATES = [ + datetime(2025, 1, 1, tzinfo=timezone.utc), + datetime(2025, 1, 2, tzinfo=timezone.utc), + datetime(2025, 1, 3, tzinfo=timezone.utc), +] + + +def seed(connection_string: str) -> None: + service = TableServiceClient.from_connection_string(connection_string) + service.create_table_if_not_exists(table_name=TABLE_NAME) + client = service.get_table_client(table_name=TABLE_NAME) + + # Clear existing data + existing = list(client.list_entities(select=["PartitionKey", "RowKey"])) + by_partition: dict[str, list[dict]] = {} + for e in existing: + pk = e["PartitionKey"] + by_partition.setdefault(pk, []).append( + {"PartitionKey": pk, "RowKey": e["RowKey"]} + ) + for pk, entities in by_partition.items(): + for i in range(0, len(entities), 100): + batch = entities[i : i + 100] + ops = [("delete", e) for e in batch] + client.submit_transaction(ops) + if existing: + print(f"Cleared {len(existing)} existing records") + + # Seed new data + entities = [] + order_idx = 0 + for date in DATES: + for user in USERS: + # 2 Recived orders per user per day + for _ in range(2): + hour = 8 + (order_idx % 12) + entities.append({ + "PartitionKey": user, + "RowKey": f"test-order-{order_idx:04d}", + "OrderTime": date.replace(hour=hour, minute=order_idx % 60), + "ReceiveTime": date.replace(hour=hour + 1, minute=order_idx % 60), + "revenue": 10.0 + order_idx, + "platformOrder": f"plat-{order_idx:04d}", + "prodCount": 1 + (order_idx % 5), + "pfId": user, + "prodId": f"prod-{order_idx % 3}", + "prodPrice": 10.0 + (order_idx % 10), + "state": 1, # Recived + "ver": 3, + "customData": None, + }) + order_idx += 1 + + # 1 Payed order per user per day (should be skipped by backup) + entities.append({ + "PartitionKey": user, + "RowKey": f"test-order-{order_idx:04d}", + "OrderTime": date.replace(hour=14, minute=30), + "ReceiveTime": None, + "revenue": 5.0, + "platformOrder": f"plat-payed-{order_idx:04d}", + "prodCount": 1, + "pfId": user, + "prodId": "prod-pay", + "prodPrice": 5.0, + "state": 0, # Payed + "ver": 3, + "customData": None, + }) + order_idx += 1 + + # Insert in batches grouped by PartitionKey + total = 0 + by_pk: dict[str, list[dict]] = {} + for e in entities: + by_pk.setdefault(e["PartitionKey"], []).append(e) + + for pk, pk_entities in by_pk.items(): + for i in range(0, len(pk_entities), 100): + batch = pk_entities[i : i + 100] + ops = [("create", e) for e in batch] + client.submit_transaction(ops) + total += len(batch) + + print(f"Seeded {total} orders across {len(by_pk)} partitions, {len(DATES)} days") + print(f" Recived (state=1): {len([e for e in entities if e['state'] == 1])}") + print(f" Payed (state=0): {len([e for e in entities if e['state'] == 0])}") + + +if __name__ == "__main__": + conn_str = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_CONNECTION_STRING + seed(conn_str) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..4dd2c78 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,170 @@ +"""Integration tests against Azure Storage emulator.""" + +import os +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from azure.data.tables import TableServiceClient + +from order_bak.backup import backup_orders +from order_bak.csv_utils import read_csv +from order_bak.delete import delete_orders +from order_bak.scan import scan_orders + +CONNECTION_STRING = os.environ.get( + "ORDER_BAK_TEST_CONNECTION_STRING", + ( + "AccountName=devstoreaccount1;" + "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/" + "K1SZFPTOtr/KBHBeksoGMGw==;" + "DefaultEndpointsProtocol=http;" + "BlobEndpoint=http://192.168.9.101:10000/devstoreaccount1;" + "QueueEndpoint=http://192.168.9.101:10001/devstoreaccount1;" + "TableEndpoint=http://192.168.9.101:10002/devstoreaccount1;" + ), +) + +TABLE_NAME = "order" + + +def _seed_orders(client, orders): + from collections import defaultdict + + by_pk: dict[str, list[dict]] = defaultdict(list) + for o in orders: + by_pk[o["PartitionKey"]].append(o) + + for pk, entities in by_pk.items(): + for i in range(0, len(entities), 100): + batch = entities[i : i + 100] + ops = [("create", e) for e in batch] + client.submit_transaction(ops) + + +def _clear_table(client): + existing = list(client.list_entities(select=["PartitionKey", "RowKey"])) + if not existing: + return + from collections import defaultdict + + by_pk: dict[str, list[dict]] = defaultdict(list) + for e in existing: + by_pk[e["PartitionKey"]].append( + {"PartitionKey": e["PartitionKey"], "RowKey": e["RowKey"]} + ) + for pk, entities in by_pk.items(): + for i in range(0, len(entities), 100): + batch = entities[i : i + 100] + ops = [("delete", e) for e in batch] + client.submit_transaction(ops) + + +@pytest.fixture +def table_client(): + service = TableServiceClient.from_connection_string(CONNECTION_STRING) + service.create_table_if_not_exists(table_name=TABLE_NAME) + client = service.get_table_client(table_name=TABLE_NAME) + _clear_table(client) + yield client + _clear_table(client) + + +@pytest.mark.integration +def test_full_pipeline(table_client, tmp_path): + # --- Seed --- + orders = [] + for day in [1, 2, 3]: + for user in ["u1", "u2"]: + # 2 Recived per user per day + for j in range(2): + orders.append({ + "PartitionKey": user, + "RowKey": f"o-d{day}-{user}-{j}", + "OrderTime": datetime(2025, 1, day, 10 + j, 0, tzinfo=timezone.utc), + "ReceiveTime": datetime(2025, 1, day, 11, 0, tzinfo=timezone.utc), + "revenue": 10.0 + day, + "platformOrder": f"plat-d{day}", + "prodCount": 1, + "pfId": user, + "prodId": "prod1", + "prodPrice": 10.0, + "state": 1, + "ver": 3, + "customData": None, + }) + # 1 Payed per user per day (should be excluded from backup) + orders.append({ + "PartitionKey": user, + "RowKey": f"o-d{day}-{user}-payed", + "OrderTime": datetime(2025, 1, day, 14, 0, tzinfo=timezone.utc), + "ReceiveTime": None, + "revenue": 5.0, + "platformOrder": "plat-p", + "prodCount": 1, + "pfId": user, + "prodId": "prod-pay", + "prodPrice": 5.0, + "state": 0, + "ver": 3, + "customData": None, + }) + _seed_orders(table_client, orders) + + # --- Scan --- + scan_dir = tmp_path / "scan" + scan_path = scan_orders( + connection_string=CONNECTION_STRING, + page_size=10, + delay_ms=0, + scan_dir=scan_dir, + ) + assert scan_path.exists() + scan_rows = list(read_csv(scan_path)) + assert len(scan_rows) == len(orders) # all orders, no filtering + + # --- Backup (Jan 1-2 only) --- + backup_dir = tmp_path / "backups" + count = backup_orders( + scan_path=scan_path, + backup_dir=backup_dir, + start=datetime(2025, 1, 1, tzinfo=timezone.utc), + end=datetime(2025, 1, 3, tzinfo=timezone.utc), + ) + # 2 days × 2 users × 2 Recived = 8 orders + assert count == 8 + + jan1 = backup_dir / "2025-01-01.csv" + jan2 = backup_dir / "2025-01-02.csv" + assert jan1.exists() + assert jan2.exists() + assert not (backup_dir / "2025-01-03.csv").exists() + + jan1_rows = list(read_csv(jan1)) + assert len(jan1_rows) == 4 # 2 users × 2 Recived + for row in jan1_rows: + assert row["state"] == 1 + + # --- Delete (Jan 1 only) --- + deleted = delete_orders( + table_client=table_client, + csv_path=jan1, + batch_size=2, + delay_ms=0, + ) + assert deleted == 4 + + # Verify Jan 1 orders are gone from table + remaining = list(table_client.list_entities()) + remaining_jan1 = [ + e for e in remaining + if e.get("OrderTime") and e["OrderTime"].day == 1 and e["state"] == 1 + ] + assert len(remaining_jan1) == 0 + + # Jan 2 Recived orders should still exist + remaining_jan2_recived = [ + e for e in remaining + if e.get("OrderTime") and e["OrderTime"].day == 2 and e["state"] == 1 + ] + assert len(remaining_jan2_recived) == 4