"""The repository's logic, tested without a database.
What is checked here is the SQL the repository DECIDES to send — which columns an upsert updates,
which rows a save deletes — so it runs in milliseconds against a connection that records statements.
Whether DataK3 then executes that SQL correctly is a separate, live test.
"""
import os
from datetime import date, datetime, timezone
import pytest
from sqlalchemy import Column, Date, Integer, MetaData, Numeric, String, Table, TIMESTAMP
from sqlalchemy.dialects import postgresql
from sqlalchemy.sql import Delete, Insert, Select
from erp_platform.repository import Aggregate, Part, upsert
md = MetaData()
quotes = Table("quote_quotes", md,
Column("quote_id", String, primary_key=True), Column("owner", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
lines = Table("quote_lines", md,
Column("quote_id", String, primary_key=True), Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True), Column("amount", Numeric(18, 2)),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
schedule = Table("quote_schedule", md,
Column("quote_id", String, primary_key=True), Column("version", Integer, primary_key=True),
Column("line_no", Integer, primary_key=True), Column("month", Date, primary_key=True),
Column("amount", Numeric(18, 2)))
contacts = Table("crm_contacts", md,
Column("id", Integer, primary_key=True), Column("name", String),
Column("phone", String), Column("created_at", TIMESTAMP), Column("created_by", String))
QUOTE = Aggregate(Part(quotes, ("quote_id",)),
Part(lines, ("quote_id", "version", "line_no")),
Part(schedule, ("quote_id", "version", "line_no", "month")))
class Recorder:
"""A connection that records what it is asked to do, and answers key reads from `stored`."""
def __init__(self, stored=None):
self.stored, self.statements = stored or {}, []
def execute(self, stmt):
self.statements.append(stmt)
if isinstance(stmt, Select):
return list(self.stored.get(stmt.get_final_froms()[0].name, set()))
return []
def of(self, kind):
return [s for s in self.statements if isinstance(s, kind)]
def sql(stmt) -> str:
return str(stmt.compile(dialect=postgresql.dialect()))
def set_clause(stmt) -> str:
text = sql(stmt)
return text.split("DO UPDATE SET", 1)[1] if "DO UPDATE SET" in text else ""
# ── upsert: a row's outcome depends on that row alone ───────────────────────────────────────
def test_a_batched_row_never_nulls_a_column_it_did_not_name():
"""The defect: padded to the batch's columns with None, row 2's phone was set to NULL."""
conn = Recorder()
upsert(conn, contacts, [{"id": 1, "name": "A", "phone": "123"}, {"id": 2, "name": "B"}], "id")
inserts = conn.of(Insert)
assert len(inserts) == 2, "rows naming different columns must be separate statements"
row2 = next(s for s in inserts if "phone" not in sql(s))
assert "phone" not in set_clause(row2)
def test_a_row_alone_and_in_a_batch_mean_the_same_thing():
alone, batched = Recorder(), Recorder()
upsert(alone, contacts, [{"id": 2, "name": "B"}], "id")
upsert(batched, contacts, [{"id": 1, "name": "A", "phone": "1"}, {"id": 2, "name": "B"}], "id")
row2 = next(s for s in batched.of(Insert) if "phone" not in sql(s))
assert set_clause(row2) == set_clause(alone.of(Insert)[0])
def test_creation_columns_are_written_once_and_never_updated():
conn = Recorder()
upsert(conn, contacts, [{"id": 1, "name": "A", "created_at": datetime(2026, 1, 1),
"created_by": "alice"}], "id")
text, sets = sql(conn.of(Insert)[0]), set_clause(conn.of(Insert)[0])
assert "created_at" in text and "created_by" in text # written on INSERT
assert "created_at" not in sets and "created_by" not in sets # never in the UPDATE
def test_an_existing_row_is_written_without_its_creation_columns():
"""DataK3 overwrites every column in VALUES whatever the SET list says (verified live), so the
only way to keep created_at is to leave it out of VALUES for a row that already exists."""
conn = Recorder({"crm_contacts": {(1,)}})
upsert(conn, contacts, [{"id": 1, "name": "A2", "created_at": datetime(2026, 6, 6)},
{"id": 2, "name": "B", "created_at": datetime(2026, 6, 6)}], "id")
by_row = {("id_m0" if "created_at" in sql(s) else "existing"): sql(s) for s in conn.of(Insert)}
assert "created_at" not in by_row["existing"] # row 1 is stored: creation kept
assert "created_at" in by_row["id_m0"] # row 2 is new: creation written
def test_save_marks_the_root_as_existing_too():
conn = Recorder({"quote_quotes": {(Q,)}})
QUOTE.save(conn, _memory([], []))
root = next(s for s in conn.of(Insert) if s.table.name == "quote_quotes")
assert "created_at" not in sql(root)
def test_a_row_with_only_keys_does_nothing_on_conflict():
conn = Recorder()
upsert(conn, contacts, [{"id": 1}], "id")
assert "DO NOTHING" in sql(conn.of(Insert)[0])
# ── save: storage ends up matching memory, at every depth ───────────────────────────────────
Q = "Q-1"
def _memory(line_nos, months):
return {"quote_quotes": [{"quote_id": Q, "owner": "alice"}],
"quote_lines": [{"quote_id": Q, "version": 1, "line_no": n, "amount": 100}
for n in line_nos],
"quote_schedule": [{"quote_id": Q, "version": 1, "line_no": n, "month": date(2026, m, 1), "amount": 50}
for n, m in months]}
def _deleted(conn) -> set:
out = set()
for d in conn.of(Delete):
text = str(d.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
out.add((d.table.name, text.split("WHERE", 1)[1].strip()))
return out
def test_save_deletes_stale_rows_at_every_depth():
"""Both of the gaps in the code these rules came from:
a schedule month trimmed from a line that STAYS, and the schedule of a line that was REMOVED."""
stored = {
"quote_lines": {(Q, 1, 1), (Q, 1, 2), (Q, 1, 3)},
"quote_schedule": {(Q, 1, 1, date(2026, 1, 1)), (Q, 1, 1, date(2026, 2, 1)),
(Q, 1, 2, date(2026, 1, 1)), (Q, 1, 3, date(2026, 1, 1))},
}
conn = Recorder(stored)
QUOTE.save(conn, _memory([1, 2], [(1, 1), (2, 1)]))
deleted = _deleted(conn)
assert len(deleted) == 3
assert any(t == "quote_lines" and "line_no = 3" in w for t, w in deleted)
assert any(t == "quote_schedule" and "line_no = 1" in w and "2026-02-01" in w for t, w in deleted), \
"a month trimmed from a kept line must be deleted"
assert any(t == "quote_schedule" and "line_no = 3" in w for t, w in deleted), \
"the schedule of a removed line must be deleted"
def test_saving_what_is_already_stored_deletes_nothing():
stored = {"quote_lines": {(Q, 1, 1)}, "quote_schedule": {(Q, 1, 1, date(2026, 1, 1))}}
conn = Recorder(stored)
QUOTE.save(conn, _memory([1], [(1, 1)]))
assert conn.of(Delete) == []
def test_every_delete_is_scoped_by_its_full_key():
"""Never a bare DELETE: this platform refuses one, and it would be the wrong instinct anyway."""
conn = Recorder({"quote_lines": {(Q, 1, 9)}, "quote_schedule": {(Q, 1, 9, date(2026, 1, 1))}})
QUOTE.save(conn, _memory([], []))
for d in conn.of(Delete):
where = sql(d).split("WHERE", 1)[1]
for col in ("quote_id", "version", "line_no"):
assert col in where
def test_reads_happen_before_any_write():
"""No read-your-writes inside a transaction: the key reads must see committed state only."""
conn = Recorder()
QUOTE.save(conn, _memory([1], [(1, 1)]))
kinds = [type(s) for s in conn.statements]
last_read = max(i for i, k in enumerate(kinds) if issubclass(k, Select))
first_write = min(i for i, k in enumerate(kinds) if not issubclass(k, Select))
assert last_read < first_write
def test_one_save_is_one_moment():
conn = Recorder()
now = datetime(2026, 10, 16, 9, 0, tzinfo=timezone.utc)
QUOTE.save(conn, _memory([1, 2], []), now=now)
for stmt in conn.of(Insert):
if "updated_at" in sql(stmt):
assert all(v == now for k, v in stmt.compile().params.items() if k.startswith("updated_at"))
def test_remove_deletes_children_first_and_always_scoped():
conn = Recorder()
QUOTE.remove(conn, {"quote_id": Q})
order = [d.table.name for d in conn.of(Delete)]
assert order == ["quote_schedule", "quote_lines", "quote_quotes"]
assert all("WHERE" in sql(d) for d in conn.of(Delete))
# ── mistakes refused at declaration ─────────────────────────────────────────────────────────
def test_a_part_whose_key_does_not_start_with_the_root_key_is_refused():
with pytest.raises(ValueError, match="must start with the root key"):
Aggregate(Part(quotes, ("quote_id",)), Part(lines, ("version", "line_no")))
def test_a_row_from_another_aggregate_is_refused():
rows = _memory([1], [])
rows["quote_lines"][0]["quote_id"] = "Q-OTHER"
with pytest.raises(ValueError, match="belongs to quote_id='Q-OTHER'"):
QUOTE.save(Recorder(), rows)
@pytest.mark.parametrize("month", ["2026-01", datetime(2026, 1, 1)])
def test_a_mistyped_key_is_refused_before_anything_is_written(month):
"""A DATE key held as a string (or a datetime) never equals what the database returns, so the
stale-row step would delete the rows the save had just written. Refused before any write."""
rows = _memory([1], [(1, 1)])
rows["quote_schedule"][0]["month"] = month
conn = Recorder()
with pytest.raises(TypeError, match="quote_schedule.month"):
QUOTE.save(conn, rows)
assert conn.of(Insert) == [] and conn.of(Delete) == []
def test_a_foreign_row_is_refused_before_anything_is_written():
rows = _memory([1], [])
rows["quote_lines"][0]["quote_id"] = "Q-OTHER"
conn = Recorder()
with pytest.raises(ValueError):
QUOTE.save(conn, rows)
assert conn.of(Insert) == []
def test_an_aggregate_has_exactly_one_root_row():
rows = _memory([1], [])
rows["quote_quotes"].append({"quote_id": Q, "owner": "bob"})
with pytest.raises(ValueError, match="exactly one quote_quotes row"):
QUOTE.save(Recorder(), rows)
# ── the platform's own db.upsert inherits the fix ───────────────────────────────────────────
def test_db_upsert_has_the_fix(monkeypatch):
monkeypatch.setenv("BUCKET", "not-used-no-connection-is-opened")
from erp_platform import db
conn = Recorder()
db.upsert(conn, contacts, [{"id": 1, "name": "A", "phone": "1"}, {"id": 2, "name": "B"}], "id")
row2 = next(s for s in conn.of(Insert) if "phone" not in sql(s))
assert "phone" not in set_clause(row2)