"""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 click.testing import CliRunner from order_bak.backup import backup_orders from order_bak.cli import main from order_bak.csv_utils import read_csv from order_bak.delete import delete_orders from order_bak.scan import scan_orders CONFIG_PATH = Path(__file__).parent.parent / "assets" / "local.toml" CONNECTION_STRING = os.environ.get( "ORDER_BAK_TEST_CONNECTION_STRING", ( "AccountName=devstoreaccount1;" "AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/" "K1SZFPTOtr/KBHBeksoGMGw==;" "DefaultEndpointsProtocol=http;" "BlobEndpoint=http://localhost:10000/devstoreaccount1;" "QueueEndpoint=http://localhost:10001/devstoreaccount1;" "TableEndpoint=http://localhost: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 (before Jan 3) --- backup_dir = tmp_path / "backups" count = backup_orders( scan_path=scan_path, backup_dir=backup_dir, cutoff=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 @pytest.mark.integration def test_delete_command_all_backups(table_client, tmp_path, monkeypatch): """CLI delete command reads all backup CSVs and deletes them from Azure.""" # Seed: 3 days × 2 users × 1 Recived order = 6 orders to back up orders = [] for day in [1, 2, 3]: for user in ["u1", "u2"]: orders.append({ "PartitionKey": user, "RowKey": f"o-d{day}-{user}", "OrderTime": datetime(2025, 1, day, 10, 0, tzinfo=timezone.utc), "ReceiveTime": datetime(2025, 1, day, 11, 0, tzinfo=timezone.utc), "revenue": 10.0, "platformOrder": "plat", "prodCount": 1, "pfId": user, "prodId": "prod1", "prodPrice": 10.0, "state": 1, "ver": 3, "customData": None, }) _seed_orders(table_client, orders) # Scan then backup (cutoff excludes nothing — all 6 are Recived before Jan 4) scan_dir = tmp_path / "scan" scan_path = scan_orders( connection_string=CONNECTION_STRING, page_size=10, delay_ms=0, scan_dir=scan_dir, ) backup_dir = tmp_path / "backups" backed_up = backup_orders( scan_path=scan_path, backup_dir=backup_dir, cutoff=datetime(2025, 1, 4, tzinfo=timezone.utc), ) assert backed_up == 6 assert len(list(backup_dir.glob("*.csv"))) == 3 # one file per day # chdir so Path("./backups") in the CLI resolves to tmp_path/backups monkeypatch.chdir(tmp_path) result = CliRunner().invoke( main, ["--config", str(CONFIG_PATH), "delete"], input="y\n", ) assert result.exit_code == 0, result.output assert "6 orders" in result.output assert "3 file" in result.output # All 6 Recived orders must be gone from the table remaining = list(table_client.list_entities()) assert len(remaining) == 0