108 lines
4.9 KiB
Python
108 lines
4.9 KiB
Python
"""Интеграционные тесты db/spec_events.py (SQLite, изолированная БД)."""
|
|
from db.connection import query
|
|
from db import spec_events, spec_current
|
|
|
|
|
|
CID = "11111111-1111-1111-1111-111111111111"
|
|
SID = "22222222-2222-2222-2222-222222222222"
|
|
DID = "33333333-3333-3333-3333-333333333333"
|
|
|
|
|
|
def _count(table, contract_id):
|
|
rows = query(f"SELECT count(*) AS c FROM {table} WHERE contract_id = %s", (contract_id,))
|
|
return rows[0]["c"]
|
|
|
|
|
|
class TestApplyOpsAdd:
|
|
def test_add(self, db):
|
|
ops = [{"action": "ADD", "new_row": {"name": "Аренда стойко-места", "price": 50000, "qty": 1, "sum": 50000, "date_start": "2025-01-01"}}]
|
|
s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {})
|
|
assert s == {"added": 1, "updated": 0, "deleted": 0, "unresolved": 0}
|
|
rows = spec_current.list_by_contract(CID)
|
|
assert len(rows) == 1
|
|
assert rows[0]["name"] == "Аренда стойко-места"
|
|
assert rows[0]["price"] == 50000
|
|
assert rows[0]["date_start"] == "2025-01-01"
|
|
|
|
def test_add_empty_name_unresolved(self, db):
|
|
ops = [{"action": "ADD", "new_row": {"name": ""}}]
|
|
s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {})
|
|
assert s == {"added": 0, "updated": 0, "deleted": 0, "unresolved": 1}
|
|
assert spec_current.list_by_contract(CID) == []
|
|
|
|
def test_add_multiple(self, db):
|
|
ops = [
|
|
{"action": "ADD", "new_row": {"name": "A", "price": 1}},
|
|
{"action": "ADD", "new_row": {"name": "B", "price": 2}},
|
|
]
|
|
s = spec_events.apply_ops(CID, SID, DID, ops, "pid", {})
|
|
assert s["added"] == 2
|
|
assert len(spec_current.list_by_contract(CID)) == 2
|
|
|
|
|
|
class TestApplyOpsUpdateDelete:
|
|
def test_update_and_delete(self, db):
|
|
spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "Аренда", "price": 50000, "qty": 1, "sum": 50000}}], "pid", {})
|
|
h = spec_events._hash("Аренда")
|
|
|
|
s = spec_events.apply_ops(CID, SID, DID, [{"action": "UPDATE", "target_hash": h, "new_values": {"price": 55000}}], "pid", {})
|
|
assert s["updated"] == 1
|
|
assert spec_current.list_by_contract(CID)[0]["price"] == 55000
|
|
|
|
s = spec_events.apply_ops(CID, SID, DID, [{"action": "DELETE", "target_hash": h}], "pid", {})
|
|
assert s["deleted"] == 1
|
|
assert spec_current.list_by_contract(CID) == []
|
|
|
|
def test_update_empty_hash_unresolved(self, db):
|
|
s = spec_events.apply_ops(CID, SID, DID, [{"action": "UPDATE", "new_values": {"price": 1}}], "pid", {})
|
|
assert s["updated"] == 0
|
|
rows = query("SELECT status FROM spec_events WHERE contract_id = %s", (CID,))
|
|
assert rows[0]["status"] == "unresolved"
|
|
|
|
|
|
class TestUnresolvedAndUnknown:
|
|
def test_unresolved_action(self, db):
|
|
ops = [{"action": "UNRESOLVED", "new_values": {"name": "X"}, "reason": "нет соответствия"}]
|
|
spec_events.apply_ops(CID, SID, DID, ops, "pid", {})
|
|
assert spec_current.list_by_contract(CID) == []
|
|
rows = query("SELECT action, status FROM spec_events WHERE contract_id = %s", (CID,))
|
|
assert rows[0]["action"] == "UNRESOLVED"
|
|
assert rows[0]["status"] == "unresolved"
|
|
|
|
def test_unknown_action(self, db):
|
|
spec_events.apply_ops(CID, SID, DID, [{"action": "WHATEVER", "new_row": {"name": "X"}}], "pid", {})
|
|
assert spec_current.list_by_contract(CID) == []
|
|
rows = query("SELECT action FROM spec_events WHERE contract_id = %s", (CID,))
|
|
assert rows[0]["action"] == "UNRESOLVED"
|
|
|
|
|
|
class TestClearAndReset:
|
|
def test_clear_current_keeps_events(self, db):
|
|
spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A", "price": 1}}], "pid", {})
|
|
assert _count("spec_current", CID) == 1
|
|
spec_events.clear_current(CID)
|
|
assert _count("spec_current", CID) == 0
|
|
assert _count("spec_events", CID) == 1 # история сохранена
|
|
|
|
def test_reset_clears_both(self, db):
|
|
spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A", "price": 1}}], "pid", {})
|
|
spec_events.reset(CID)
|
|
assert _count("spec_current", CID) == 0
|
|
assert _count("spec_events", CID) == 0
|
|
|
|
|
|
class TestHashAndSeq:
|
|
def test_hash_normalizes(self):
|
|
assert spec_events._hash(" Арена стойко-места ") == spec_events._hash("арена стойко-места")
|
|
|
|
def test_hash_includes_date(self):
|
|
assert spec_events._hash("Аренда", "01.01.2025") != spec_events._hash("Аренда", "01.02.2025")
|
|
|
|
def test_hash_len(self):
|
|
assert len(spec_events._hash("x")) == 16
|
|
|
|
def test_seq(self, db):
|
|
assert spec_events.get_next_seq(CID) == 1
|
|
spec_events.apply_ops(CID, SID, DID, [{"action": "ADD", "new_row": {"name": "A"}}], "pid", {})
|
|
assert spec_events.get_next_seq(CID) == 2
|