order-bak/tests/test_integration.py

"""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