131 lines
4.0 KiB
Python
131 lines
4.0 KiB
Python
from datetime import datetime, timezone
|
|
from order_bak.csv_utils import write_csv
|
|
from order_bak.backup import backup_orders
|
|
|
|
|
|
def _make_entity(pk, rk, order_time, state, receive_time=None):
|
|
return {
|
|
"PartitionKey": pk,
|
|
"RowKey": rk,
|
|
"OrderTime": order_time,
|
|
"ReceiveTime": receive_time,
|
|
"revenue": 10.0,
|
|
"platformOrder": "plat1",
|
|
"prodCount": 1,
|
|
"pfId": pk,
|
|
"prodId": "prod1",
|
|
"prodPrice": 10.0,
|
|
"state": state,
|
|
"ver": 3,
|
|
"customData": None,
|
|
}
|
|
|
|
|
|
def test_backup_filters_by_state_and_date(tmp_path):
|
|
scan_dir = tmp_path / "scan"
|
|
backup_dir = tmp_path / "backups"
|
|
scan_dir.mkdir()
|
|
backup_dir.mkdir()
|
|
|
|
entities = [
|
|
_make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
|
|
_make_entity("u2", "o2", datetime(2025, 1, 1, 12, tzinfo=timezone.utc), 0), # Payed, skip
|
|
_make_entity("u3", "o3", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
|
|
_make_entity("u4", "o4", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1), # at cutoff, skip
|
|
]
|
|
scan_csv = scan_dir / "scan-2025-01-03.csv"
|
|
write_csv(scan_csv, entities)
|
|
|
|
result = backup_orders(
|
|
scan_path=scan_csv,
|
|
backup_dir=backup_dir,
|
|
cutoff=datetime(2025, 1, 3, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert result == 2
|
|
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()
|
|
|
|
from order_bak.csv_utils import read_csv
|
|
jan1_rows = list(read_csv(jan1))
|
|
assert len(jan1_rows) == 1
|
|
assert jan1_rows[0]["RowKey"] == "o1"
|
|
|
|
|
|
def test_backup_filters_by_cutoff(tmp_path):
|
|
scan_dir = tmp_path / "scan"
|
|
backup_dir = tmp_path / "backups"
|
|
scan_dir.mkdir()
|
|
backup_dir.mkdir()
|
|
|
|
entities = [
|
|
_make_entity("u1", "o1", datetime(2025, 1, 1, 10, tzinfo=timezone.utc), 1),
|
|
_make_entity("u2", "o2", datetime(2025, 1, 2, 8, tzinfo=timezone.utc), 1),
|
|
_make_entity("u3", "o3", datetime(2025, 1, 3, 8, tzinfo=timezone.utc), 1), # at/after cutoff
|
|
]
|
|
scan_csv = scan_dir / "scan-2025-01-03.csv"
|
|
write_csv(scan_csv, entities)
|
|
|
|
cutoff = datetime(2025, 1, 3, tzinfo=timezone.utc)
|
|
result = backup_orders(
|
|
scan_path=scan_csv,
|
|
backup_dir=backup_dir,
|
|
cutoff=cutoff,
|
|
)
|
|
|
|
assert result == 2
|
|
assert (backup_dir / "2025-01-01.csv").exists()
|
|
assert (backup_dir / "2025-01-02.csv").exists()
|
|
assert not (backup_dir / "2025-01-03.csv").exists()
|
|
|
|
|
|
def test_backup_cutoff_at_midnight(tmp_path):
|
|
"""Order exactly at cutoff midnight should be excluded."""
|
|
scan_dir = tmp_path / "scan"
|
|
backup_dir = tmp_path / "backups"
|
|
scan_dir.mkdir()
|
|
backup_dir.mkdir()
|
|
|
|
entities = [
|
|
_make_entity("u1", "o1", datetime(2025, 1, 3, 0, 0, 0, tzinfo=timezone.utc), 1), # exactly at cutoff
|
|
_make_entity("u2", "o2", datetime(2025, 1, 2, 23, 59, 59, tzinfo=timezone.utc), 1), # just before
|
|
]
|
|
scan_csv = scan_dir / "scan.csv"
|
|
write_csv(scan_csv, entities)
|
|
|
|
result = backup_orders(
|
|
scan_path=scan_csv,
|
|
backup_dir=backup_dir,
|
|
cutoff=datetime(2025, 1, 3, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert result == 1
|
|
assert (backup_dir / "2025-01-02.csv").exists()
|
|
assert not (backup_dir / "2025-01-03.csv").exists()
|
|
|
|
|
|
def test_backup_empty_result(tmp_path):
|
|
"""No orders before cutoff returns 0 and no files."""
|
|
scan_dir = tmp_path / "scan"
|
|
backup_dir = tmp_path / "backups"
|
|
scan_dir.mkdir()
|
|
backup_dir.mkdir()
|
|
|
|
entities = [
|
|
_make_entity("u1", "o1", datetime(2025, 1, 5, 10, tzinfo=timezone.utc), 1),
|
|
]
|
|
scan_csv = scan_dir / "scan.csv"
|
|
write_csv(scan_csv, entities)
|
|
|
|
result = backup_orders(
|
|
scan_path=scan_csv,
|
|
backup_dir=backup_dir,
|
|
cutoff=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert result == 0
|
|
assert list(backup_dir.glob("*.csv")) == []
|