105 lines
4.4 KiB
Python
105 lines
4.4 KiB
Python
"""Upload duplicate protection — the historical double-count bug.
|
|
|
|
Re-uploading a file with the same name used to overwrite it on disk but insert a SECOND
|
|
session_files row pointing at the same path, so processing parsed and summed the file
|
|
twice. Same content under a different name was equally unguarded."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.main import app
|
|
from app.db import models
|
|
from app.db.database import SessionLocal, init_db
|
|
from tests.test_excel_export import make_amazon_xlsx
|
|
|
|
_TMP = tempfile.mkdtemp(prefix="ar_dedup_test_")
|
|
|
|
|
|
def _upload(c, sid: int, path: str, as_name: str | None = None):
|
|
with open(path, "rb") as fh:
|
|
return c.post(f"/api/sessions/{sid}/files",
|
|
files={"files": (as_name or os.path.basename(path), fh)})
|
|
|
|
|
|
def _file_rows(sid: int) -> list[models.SessionFile]:
|
|
db = SessionLocal()
|
|
try:
|
|
return db.query(models.SessionFile).filter_by(session_id=sid).all()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_same_filename_reupload_updates_row_not_duplicates():
|
|
init_db()
|
|
a = os.path.join(_TMP, "USA jan.xlsx")
|
|
make_amazon_xlsx(a, order_rows=6)
|
|
with TestClient(app) as c:
|
|
sid = c.post("/api/sessions", json={"name": "dedup-name", "month_end_date": "2027-01-31",
|
|
"allow_duplicate": True}).json()["id"]
|
|
first = _upload(c, sid, a).json()
|
|
assert len(first["files"]) == 1 and first["skipped"] == []
|
|
first_id = first["files"][0]["id"]
|
|
|
|
# Different bytes, same filename -> the existing row is replaced, never doubled.
|
|
b = os.path.join(_TMP, "USA jan v2.xlsx")
|
|
make_amazon_xlsx(b, order_rows=9)
|
|
second = _upload(c, sid, b, as_name="USA jan.xlsx").json()
|
|
assert len(second["files"]) == 1 and second["skipped"] == []
|
|
assert second["files"][0]["id"] == first_id # updated in place
|
|
|
|
rows = _file_rows(sid)
|
|
assert len(rows) == 1
|
|
assert rows[0].sha256 == second["files"][0]["sha256"]
|
|
|
|
|
|
def test_identical_bytes_same_name_is_skipped():
|
|
init_db()
|
|
a = os.path.join(_TMP, "USA feb.xlsx")
|
|
make_amazon_xlsx(a, order_rows=6)
|
|
with TestClient(app) as c:
|
|
sid = c.post("/api/sessions", json={"name": "dedup-same", "month_end_date": "2027-02-28",
|
|
"allow_duplicate": True}).json()["id"]
|
|
assert _upload(c, sid, a).status_code == 200
|
|
again = _upload(c, sid, a).json()
|
|
assert again["files"] == []
|
|
assert len(again["skipped"]) == 1
|
|
assert "unchanged" in again["skipped"][0]["reason"]
|
|
assert len(_file_rows(sid)) == 1
|
|
|
|
|
|
def test_identical_content_under_new_name_is_skipped():
|
|
init_db()
|
|
a = os.path.join(_TMP, "USA mar.xlsx")
|
|
make_amazon_xlsx(a, order_rows=6)
|
|
with TestClient(app) as c:
|
|
sid = c.post("/api/sessions", json={"name": "dedup-bytes", "month_end_date": "2027-03-31",
|
|
"allow_duplicate": True}).json()["id"]
|
|
assert _upload(c, sid, a).status_code == 200
|
|
renamed = _upload(c, sid, a, as_name="USA mar COPY.xlsx").json()
|
|
assert renamed["files"] == []
|
|
assert "already uploaded as" in renamed["skipped"][0]["reason"]
|
|
assert len(_file_rows(sid)) == 1
|
|
|
|
|
|
def test_double_upload_no_longer_doubles_the_totals():
|
|
"""End to end: upload, process, re-upload the SAME file, re-process — totals unchanged."""
|
|
init_db()
|
|
a = os.path.join(_TMP, "USA apr.xlsx")
|
|
make_amazon_xlsx(a, order_rows=8)
|
|
with TestClient(app) as c:
|
|
sid = c.post("/api/sessions", json={"name": "dedup-e2e", "month_end_date": "2027-04-30",
|
|
"allow_duplicate": True}).json()["id"]
|
|
assert _upload(c, sid, a).status_code == 200
|
|
assert c.post(f"/api/sessions/{sid}/process").status_code == 200
|
|
assert c.get(f"/api/sessions/{sid}/status").json()["status"] == "processed"
|
|
before = c.get(f"/api/sessions/{sid}/summary").json()["closing_receivable_usd"]
|
|
|
|
assert _upload(c, sid, a).status_code == 200 # skipped as unchanged
|
|
assert c.post(f"/api/sessions/{sid}/process").status_code == 200
|
|
assert c.get(f"/api/sessions/{sid}/status").json()["status"] == "processed"
|
|
after = c.get(f"/api/sessions/{sid}/summary").json()["closing_receivable_usd"]
|
|
assert after == before
|