Importer assessor key · open after the independent attempt

from .model import page
from .transport import fetch


def run(server, clock, store, budget=5.0, max_pages=100, after_persist=lambda: None):
    if budget <= 0 or max_pages <= 0:
        raise ValueError("positive budgets required")
    deadline = clock.now + budget
    cursor, done = store.checkpoint()
    seen = set()
    if done:
        return store.rows()
    for _ in range(max_pages):
        if cursor in seen:
            raise ValueError("cursor cycle")
        seen.add(cursor)
        rows, next_cursor = page(fetch(server, clock, cursor, deadline))
        if clock.now >= deadline:
            raise TimeoutError("total deadline before persistence")
        if next_cursor is not None and next_cursor in seen:
            raise ValueError("cursor cycle")
        store.persist(rows)
        after_persist()  # deliberate crash window: restart must safely replay this page
        if clock.now >= deadline:
            raise TimeoutError("total deadline after persistence; replay on restart")
        store.advance(next_cursor)
        if next_cursor is None:
            return store.rows()
        cursor = next_cursor
    raise ValueError("page budget exhausted")

"""Injected transport owns timeout enforcement; importer owns the total budget."""
class FakeClock:
    def __init__(self):
        self.now = 0.0
        self.waits = []

    def sleep(self, seconds):
        self.waits.append(seconds)
        self.now += seconds


class FakeServer:
    def __init__(self, responses, clock):
        self.responses, self.clock, self.calls = responses, clock, []
        self.counts = {}

    def get(self, cursor, timeout):
        key = cursor if cursor is not None else "START"
        self.calls.append((key, self.clock.now, timeout))
        count = self.counts.get(key, 0)
        options = self.responses[key]
        response = options[min(count, len(options) - 1)]
        self.counts[key] = count + 1
        elapsed = response.get("elapsed", 0)
        self.clock.now += min(elapsed, timeout)
        if elapsed >= timeout:
            raise TimeoutError("request deadline")
        return response


def fetch(server, clock, cursor, deadline, attempts=3):
    for attempt in range(attempts):
        remaining = deadline - clock.now
        if remaining <= 0:
            raise TimeoutError("total deadline")
        response = server.get(cursor, timeout=remaining)
        if clock.now >= deadline:
            raise TimeoutError("total deadline")
        status = response["status"]
        if status == 200:
            return response["body"]
        if status not in (429, 500, 502, 503, 504):
            raise ValueError(f"non-retryable HTTP {status}")
        if attempt == attempts - 1:
            raise TimeoutError("attempt budget exhausted")
        # This exercise's API specifies numeric seconds, not HTTP-date syntax.
        delay = response.get("retryAfter", 0.1 * 2**attempt)
        if isinstance(delay, bool) or not isinstance(delay, (int, float)) or not 0 <= delay < float("inf"):
            raise ValueError("invalid Retry-After")
        if clock.now + delay >= deadline:
            raise TimeoutError("retry exceeds total deadline")
        clock.sleep(delay)
    raise AssertionError("unreachable")

"""Domain validation: cents are integers; malformed pages have no partial effect."""
from decimal import Decimal, InvalidOperation


def normalize(item):
    if not isinstance(item, dict) or not isinstance(item.get("id"), str) or not item["id"]:
        raise ValueError("id required")
    if item.get("currency") != "USD" or not isinstance(item.get("amount"), str) or len(item["amount"]) > 128:
        raise ValueError("USD amount must be a decimal string")
    try:
        amount = Decimal(item["amount"])
        if not amount.is_finite():
            raise ValueError("whole cents required")
        if amount.copy_abs() > Decimal("10000000000"):
            raise ValueError("amount out of range")
        # Never multiply under Decimal's ambient context: it can round away a tiny
        # fractional cent. The parsed coefficient and exponent are exact.
        sign, digits, exponent = amount.as_tuple()
        coefficient = 0
        for digit in digits:
            coefficient = coefficient * 10 + digit
        if not coefficient:
            cents = 0
        elif exponent >= -2:
            cents = coefficient * 10 ** (exponent + 2)
        else:
            places = -exponent - 2
            if places > len(digits):
                raise ValueError("whole cents required")
            cents, fraction = divmod(coefficient, 10 ** places)
            if fraction:
                raise ValueError("whole cents required")
        return (item["id"], -cents if sign else cents, "USD")
    except InvalidOperation as exc:
        raise ValueError("invalid amount") from exc


def page(body):
    if not isinstance(body, dict) or not isinstance(body.get("items"), list):
        raise ValueError("items list required")
    if "nextCursor" not in body or (body["nextCursor"] is not None and
                                      (not isinstance(body["nextCursor"], str) or not body["nextCursor"])):
        raise ValueError("opaque cursor or null required")
    return [normalize(item) for item in body["items"]], body["nextCursor"]

import sqlite3


class Store:
    def __init__(self, path):
        self.db = sqlite3.connect(path)
        self.db.executescript("""
          CREATE TABLE IF NOT EXISTS transactions(id TEXT PRIMARY KEY, cents INTEGER NOT NULL, currency TEXT NOT NULL);
          CREATE TABLE IF NOT EXISTS progress(singleton INTEGER PRIMARY KEY CHECK(singleton=1), cursor TEXT, done INTEGER NOT NULL);
          INSERT OR IGNORE INTO progress VALUES(1,NULL,0);
        """)

    def checkpoint(self):
        return self.db.execute("SELECT cursor,done FROM progress WHERE singleton=1").fetchone()

    def persist(self, rows):
        with self.db:
            for row in rows:
                old = self.db.execute("SELECT id,cents,currency FROM transactions WHERE id=?", (row[0],)).fetchone()
                if old is not None and old != row:
                    raise ValueError("duplicate id with conflicting payload")
                self.db.execute("INSERT OR IGNORE INTO transactions VALUES(?,?,?)", row)

    def advance(self, cursor):
        with self.db:
            self.db.execute("UPDATE progress SET cursor=?,done=? WHERE singleton=1", (cursor, cursor is None))

    def rows(self):
        return self.db.execute("SELECT id,cents,currency FROM transactions ORDER BY id").fetchall()

    def close(self):
        self.db.close()

"""Assessor-only variants. Select candidate package after its baseline is repaired."""
import argparse
import importlib
from pathlib import Path
import sys
import tempfile
import unittest

parser = argparse.ArgumentParser()
parser.add_argument("--package", choices=["reference", "starter"], default="reference")
args, remaining = parser.parse_known_args()
root = Path(__file__).resolve().parents[2] / "curriculum/02-applications/04-testing/labs/importer"
sys.path.insert(0, str(root))
run = importlib.import_module(args.package + ".importer").run
Store = importlib.import_module(args.package + ".store").Store
transport = importlib.import_module(args.package + ".transport")


def item(ident="held", amount="0.29"):
    return {"id":ident, "amount":amount, "currency":"USD"}


def page(items, cursor=None):
    return {"status":200, "body":{"items":items, "nextCursor":cursor}}


class HeldBack(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.path = str(Path(self.temp.name) / "held.db")
        self.store = Store(self.path)
        self.clock = transport.FakeClock()

    def tearDown(self):
        self.store.close()
        self.temp.cleanup()

    def test_multi_hop_cycle(self):
        server = transport.FakeServer({"START":[page([],"p2")],"p2":[page([],"p3")],"p3":[page([],"p2")]},self.clock)
        with self.assertRaisesRegex(ValueError,"cycle"):
            run(server,self.clock,self.store)
        self.assertEqual(len(server.calls),3)

    def test_retry_consumes_transport_budget(self):
        server = transport.FakeServer({"START":[{"status":429,"retryAfter":0.8,"elapsed":0.3}]},self.clock)
        with self.assertRaises(TimeoutError):
            run(server,self.clock,self.store,budget=1)
        self.assertEqual(len(server.calls),1)
        self.assertEqual(self.clock.waits,[])

    def test_crash_and_equal_decimal_spelling(self):
        server = transport.FakeServer({"START":[page([item(amount="0.290")])]},self.clock)
        def crash():
            raise RuntimeError("lost process")
        with self.assertRaises(RuntimeError):
            run(server,self.clock,self.store,after_persist=crash)
        self.store.close()
        self.store = Store(self.path)
        server = transport.FakeServer({"START":[page([item(amount="0.29")])]},self.clock)
        self.assertEqual(run(server,self.clock,self.store),[("held",29,"USD")])

    def test_late_malformed_page_preserves_progress(self):
        server = transport.FakeServer({"START":[page([item("first")],"next")],
                                      "next":[page([item("valid"),item("bad","0.00001")])]},self.clock)
        with self.assertRaises(ValueError):
            run(server,self.clock,self.store)
        self.assertEqual(self.store.rows(),[("first",29,"USD")])
        self.assertEqual(self.store.checkpoint(),("next",0))


if __name__ == "__main__":
    unittest.main(argv=[sys.argv[0]] + remaining)