"""Run with Python 3. Synthetic SSIS migration contract using local SQLite."""

import csv
import hashlib
import sqlite3
from contextlib import closing
from decimal import Decimal, InvalidOperation
from io import StringIO
from pathlib import Path
from tempfile import TemporaryDirectory


class SimulatedStop(RuntimeError):
    pass


def load(db, name, contents, regions, stop=None):
    checksum = hashlib.sha256(contents.encode()).hexdigest()
    with db:
        receipt = db.execute(
            "SELECT checksum FROM receipts WHERE file = ?", (name,)
        ).fetchone()
        if receipt:
            if receipt != (checksum,):
                raise ValueError("Changed file needs an approved correction")
            return "receipt reused"

        for ordinal, (key, region, amount) in enumerate(csv.reader(StringIO(contents)), 1):
            matches = regions.get(region, ())
            reason = {0: "lookup miss", 1: None}.get(len(matches), "multiple matches")
            if reason is None:
                try:
                    cents = Decimal(amount) * 100
                    if not cents.is_finite() or cents != cents.to_integral_value():
                        raise InvalidOperation
                except InvalidOperation:
                    reason = "invalid amount"
            if reason:
                db.execute("INSERT INTO rejects VALUES (?, ?, ?, ?)", (name, ordinal, key, reason))
            else:
                db.execute(
                    "INSERT INTO accepted VALUES (?, ?, ?, ?, ?)",
                    (name, ordinal, key, matches[0], int(cents)),
                )

        db.execute("INSERT INTO receipts VALUES (?, ?)", (name, checksum))
        if stop == "before commit":
            raise SimulatedStop(stop)
    if stop == "after commit":
        raise SimulatedStop(stop)
    return "committed"


def interrupt(db, name, contents, regions, phase):
    try:
        load(db, name, contents, regions, stop=phase)
    except SimulatedStop as error:
        assert str(error) == phase
    else:
        raise AssertionError(f"Expected a stop {phase}")


def main():
    with TemporaryDirectory() as directory:
        run_lab(Path(directory) / "target.sqlite")


def run_lab(path):
    files = {
        "A.csv": "A1,R1,18.25\nA2,unknown,1.25\n",
        "B.csv": "B1,R1,1.75\nB2,R1,invalid\nB3,R2,5.00\n",
    }
    regions = {"R1": ("West",), "R2": ("East", "North")}
    with closing(sqlite3.connect(path)) as db:
        db.executescript(
            """
            CREATE TABLE accepted (
              file TEXT, ordinal INTEGER, key TEXT, region TEXT, cents INTEGER,
              PRIMARY KEY (file, ordinal)
            );
            CREATE TABLE rejects (
              file TEXT, ordinal INTEGER, key TEXT, reason TEXT,
              PRIMARY KEY (file, ordinal)
            );
            CREATE TABLE receipts (file TEXT PRIMARY KEY, checksum TEXT NOT NULL);
            """
        )
        interrupt(db, "A.csv", files["A.csv"], regions, "before commit")

    with closing(sqlite3.connect(path)) as db:
        for table in ("accepted", "rejects", "receipts"):
            assert db.execute(f"SELECT COUNT(*) FROM {table}").fetchone() == (0,)
        assert load(db, "A.csv", files["A.csv"], regions) == "committed"
        interrupt(db, "B.csv", files["B.csv"], regions, "after commit")

    with closing(sqlite3.connect(path)) as db:
        assert load(db, "B.csv", files["B.csv"], regions) == "receipt reused"

        # This fixture oracle is separate from the replacement implementation.
        assert db.execute("SELECT * FROM accepted ORDER BY file, ordinal").fetchall() == [
            ("A.csv", 1, "A1", "West", 1825),
            ("B.csv", 1, "B1", "West", 175),
        ]
        assert db.execute("SELECT * FROM rejects ORDER BY file, ordinal").fetchall() == [
            ("A.csv", 2, "A2", "lookup miss"),
            ("B.csv", 2, "B2", "invalid amount"),
            ("B.csv", 3, "B3", "multiple matches"),
        ]
        assert db.execute("SELECT file FROM receipts ORDER BY file").fetchall() == [
            ("A.csv",), ("B.csv",),
        ]
        try:
            load(db, "A.csv", files["A.csv"] + "A3,R1,0.50\n", regions)
        except ValueError as error:
            assert str(error) == "Changed file needs an approved correction"
        else:
            raise AssertionError("A changed file should require an approved correction")

    print("Parity passed: 2 accepted rows (2000 cents), 3 rejects and 2 receipts.")
    print("The pre-commit retry wrote once. The post-commit retry reused the receipt.")


if __name__ == "__main__":
    main()