让 Codex 分摊 100 元：先写清总额守恒与余数归属

"""Explicit acceptance tests: python -m unittest -v test_allocate."""
import copy
from fractions import Fraction
import itertools
import json
from pathlib import Path
import unittest
from allocate import allocate

ROOT = Path(__file__).resolve().parent


class AllocationTests(unittest.TestCase):
    def assert_invariants(self, total, rows):
        before = copy.deepcopy(rows)
        result = allocate(total, rows)
        self.assertEqual(rows, before)
        self.assertEqual([identifier for identifier, _ in result], sorted(r[0] for r in rows))
        self.assertEqual(sum(amount for _, amount in result), total)
        weights = dict(rows)
        denominator = sum(weights.values())
        for identifier, amount in result:
            self.assertIs(type(amount), int)
            self.assertGreaterEqual(amount, 0)
            quota = Fraction(total * weights[identifier], denominator)
            floor = quota.numerator // quota.denominator
            ceiling = -(-quota.numerator // quota.denominator)
            self.assertIn(amount, (floor, ceiling))
            if weights[identifier] == 0:
                self.assertEqual(amount, 0)
        return result

    def assert_code(self, code, total, rows):
        before = copy.deepcopy(rows)
        with self.assertRaises(ValueError) as caught:
            allocate(total, rows)
        self.assertEqual(str(caught.exception), code)
        self.assertEqual(rows, before)

    def test_fixtures(self):
        cases = json.loads((ROOT / "fixtures-input.json").read_text())
        expected = json.loads((ROOT / "fixtures-output.json").read_text())
        self.assertEqual(len(cases), len(expected))
        for case, output in zip(cases, expected):
            with self.subTest(case=case):
                if "error" in output:
                    self.assert_code(output["error"], case["total_minor"], case["rows"])
                else:
                    result = self.assert_invariants(case["total_minor"], case["rows"])
                    self.assertEqual([list(pair) for pair in result], output["result"])

    def test_all_three_row_fixture_permutations(self):
        fixtures = [(10000, [("C", 1), ("A", 1), ("B", 1)], [("A",3334),("B",3333),("C",3333)]),
                    (7, [("A",5),("B",3),("C",2)], [("A",4),("B",2),("C",1)]),
                    (2, [("C",1),("B",1),("A",1)], [("A",1),("B",1),("C",0)])]
        for total, rows, expected in fixtures:
            for permutation in itertools.permutations(rows):
                result = self.assert_invariants(total, list(permutation))
                self.assertEqual(result, expected)
                self.assertEqual(json.dumps(result), json.dumps(expected))

    def test_exhaustive_small_cases(self):
        # 31 * (6**3 - 1) = 6,665 exact-quota and conservation checks.
        for weights in itertools.product(range(6), repeat=3):
            if not any(weights):
                continue
            rows = list(zip(("A", "B", "C"), weights))
            for total in range(31):
                self.assert_invariants(total, rows)

    def test_collection_contract(self):
        for rows in (None, {}, (), (("A", 1),), "A", [], [["A"]], [["A",1,2]], [None], ["A1"], [{"A":1}]):
            self.assert_code("INVALID_ROWS", 1, rows)
        self.assert_code("INVALID_ROWS", 1, [(f"A{i}", 1) for i in range(1001)])
        self.assertEqual(allocate(2, [["B", 1], ("A", 1)]), [("A",1),("B",1)])
        rows = [(f"A{i}", 1) for i in range(1000)]
        self.assert_invariants(10**12, rows)
        with self.assertRaisesRegex(ValueError, "^INVALID_ROWS$"):
            allocate(1, iter([("A", 1)]))

    def test_invalid_ids(self):
        for identifier in ("a", "", 0, True, None, " A", "A ", "A\n", "Ａ", "Aé", "A-1", "1A", "A"*33):
            self.assert_code("INVALID_ID", 1, [(identifier, 1)])
        self.assertEqual(allocate(1, [("A" + "_"*31, 1)]), [("A" + "_"*31, 1)])
        self.assert_code("DUPLICATE_ID", 1, [("A",1),("A",2)])

    def test_exact_integer_types(self):
        class IntSubclass(int):
            pass
        for total in (True, False, 1.0, "1", None, -1, 10**12+1, IntSubclass(1)):
            self.assert_code("INVALID_TOTAL", total, [("A",1)])
        for weight in (True, False, 1.0, "1", None, -1, 10**9+1, IntSubclass(1)):
            self.assert_code("INVALID_WEIGHT", 1, [("A",weight)])

    def test_exact_container_and_id_types(self):
        class ListSubclass(list):
            pass
        class TupleSubclass(tuple):
            pass
        class StringSubclass(str):
            pass
        self.assert_code("INVALID_ROWS", 1, ListSubclass([("A",1)]))
        self.assert_code("INVALID_ROWS", 1, [ListSubclass(["A",1])])
        self.assert_code("INVALID_ROWS", 1, [TupleSubclass(("A",1))])
        self.assert_code("INVALID_ID", 1, [(StringSubclass("A"),1)])

    def test_zero_cases(self):
        for total in (0, 1):
            self.assert_code("ZERO_WEIGHT_SUM", total, [("A",0)])
        self.assertEqual(allocate(0, [("B",2),("A",0)]), [("A",0),("B",0)])
        self.assertEqual(allocate(1, [("A",0),("B",1)]), [("A",0),("B",1)])

    def test_validation_precedence(self):
        # Intentionally multiple faults: these specifically test precedence.
        self.assert_code("INVALID_TOTAL", True, [])
        self.assert_code("INVALID_ROWS", 1, [["a"]])
        self.assert_code("INVALID_ID", 1, [("a",-1)])
        self.assert_code("DUPLICATE_ID", 1, [("A",1),("A",-1)])
        self.assert_code("INVALID_WEIGHT", 1, [("A",-1),("a",1)])
        self.assert_code("INVALID_ID", 0, [("a",0)])

    def test_zero_total_still_validates_late_rows(self):
        self.assert_code("INVALID_WEIGHT", 0, [["B",2],["A",True]])
        self.assert_code("DUPLICATE_ID", 0, [["B",2],["B",1]])

    def test_large_exact_products(self):
        self.assert_invariants(10**12-1, [("C",10**9),("A",10**9-1),("B",1)])

    def test_exact_remainder_precision_trap(self):
        self.assertEqual(
            allocate(999999999501, [("A",999999999),("B",1000000000)]),
            [("A",499999999500),("B",500000000001)])

    def test_total_increase_can_reduce_a_row(self):
        rows = [("A",5),("B",3),("C",1)]
        self.assertEqual(allocate(4, rows), [("A",2),("B",1),("C",1)])
        self.assertEqual(allocate(5, rows), [("A",3),("B",2),("C",0)])


if __name__ == "__main__":
    unittest.main()

"""Print deterministic fixture outputs; no service or coding-agent calls."""
import json
from pathlib import Path
from allocate import allocate


def evaluate(case):
    try:
        return {"result": allocate(case["total_minor"], case["rows"])}
    except ValueError as error:
        return {"error": str(error)}


if __name__ == "__main__":
    cases = json.loads(Path(__file__).with_name("fixtures-input.json").read_text())
    print(json.dumps([evaluate(case) for case in cases], indent=2))

"""Generate committed synthetic fixtures from explicit expected answers.

This authoring helper does not import or call the implementation under test.
"""
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parent
cases = []
outputs = []


def add(total, rows, result=None, error=None):
    cases.append({"total_minor": total, "rows": rows})
    outputs.append({"error": error} if error else {"result": result})


add(10000, [["C",1],["A",1],["B",1]], [["A",3334],["B",3333],["C",3333]])
add(7, [["A",5],["B",3],["C",2]], [["A",4],["B",2],["C",1]])
add(2, [["C",1],["B",1],["A",1]], [["A",1],["B",1],["C",0]])
add(5, [["B",2],["A",0]], [["A",0],["B",5]])
add(0, [["B",2],["A",1]], [["A",0],["B",0]])
add(11, [["ONLY",7]], [["ONLY",11]])
add(10**12, [["A",10**9],["B",10**9]], [["A",500000000000],["B",500000000000]])
for total in (-1, True, 1.0, 10**12+1):
    add(total, [["A",1]], error="INVALID_TOTAL")
add(1, [], error="INVALID_ROWS")
add(1, [[f"A{i}",1] for i in range(1001)], error="INVALID_ROWS")
for identifier in ("a", "", 0):
    add(1, [[identifier,1]], error="INVALID_ID")
add(1, [["A",1],["A",2]], error="DUPLICATE_ID")
for weight in (-1, True, 1.5, "1", 10**9+1):
    add(1, [["A",weight]], error="INVALID_WEIGHT")
for total in (0,1):
    add(total, [["A",0]], error="ZERO_WEIGHT_SUM")
for filename, data in (("fixtures-input.json", cases), ("fixtures-output.json", outputs)):
    (ROOT / filename).write_text(json.dumps(data, indent=2) + "\n")

"""Capture actual baseline, deterministic demo, and isolated mutation results.

Each mutant must compile, be imported from its own isolated directory, fail
acceptance by assertion failure, and produce no test errors/import failures.
"""
import hashlib
import json
import os
from pathlib import Path
import platform
import shutil
import subprocess
import sys
import tempfile
import time

ROOT = Path(__file__).resolve().parent
LOGS = ROOT / "verification"
LOGS.mkdir(exist_ok=True)
SOURCE = (ROOT / "allocate.py").read_text()


def replace_once(old, new):
    if SOURCE.count(old) != 1:
        raise AssertionError(f"Mutation anchor must occur once: {old!r}")
    return SOURCE.replace(old, new, 1)


mutations = {
    "independent_rounding": replace_once(
        "    return sorted(amounts.items())",
        "    return sorted((identifier, round(total_minor * weight / sum_weights))\n"
        "                  for identifier, weight in validated)"),
    "dump_residual_into_last_row": replace_once(
        "    for identifier, _ in ranked[:remaining]:\n        amounts[identifier] += 1",
        "    amounts[validated[-1][0]] += remaining"),
    "input_order_tie_ownership": replace_once(
        "key=lambda item: (-item[1], item[0])", "key=lambda item: -item[1]"),
    "accept_boolean_total": replace_once(
        "type(total_minor) is not int", "not isinstance(total_minor, int)"),
    "accept_boolean_weight": replace_once(
        "type(weight) is not int", "not isinstance(weight, int)"),
    "silently_merge_duplicate_ids": replace_once(
        "        if identifier in seen:\n            raise ValueError(\"DUPLICATE_ID\")",
        "        if identifier in seen:\n"
        "            validated = [(key, value + weight if key == identifier else value)\n"
        "                         for key, value in validated]\n"
        "            continue"),
}


def run(command, cwd, name):
    start = time.perf_counter()
    env = os.environ.copy()
    env.pop("PYTHONPATH", None)
    process = subprocess.run(command, cwd=cwd, env=env, text=True,
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    duration = time.perf_counter() - start
    (LOGS / (name + ".stdout.txt")).write_text(process.stdout)
    (LOGS / (name + ".stderr.txt")).write_text(process.stderr)
    return process, {"command": command, "exit_code": process.returncode,
                     "seconds": duration, "stdout": name + ".stdout.txt",
                     "stderr": name + ".stderr.txt"}


report = {"python": sys.version, "executable": sys.executable,
          "platform": platform.platform(),
          "source_sha256": hashlib.sha256(SOURCE.encode()).hexdigest(),
          "small_invariant_cases": 6665, "fixture_cases": 24, "mutations": {}}
baseline, report["baseline"] = run([sys.executable, "-m", "unittest", "-v", "test_allocate"], ROOT, "baseline")
if baseline.returncode:
    raise SystemExit("Baseline failed; see verification/baseline.stderr.txt")
demo, report["demo"] = run([sys.executable, "demo.py"], ROOT, "demo")
expected = (ROOT / "fixtures-output.json").read_text()
report["demo"]["normalized_text_equal_expected"] = demo.stdout == expected
if demo.returncode or demo.stdout != expected:
    raise SystemExit("Demo differs from committed expected output")

for name, source in mutations.items():
    compile(source, "allocate.py", "exec")
    (LOGS / (name + ".py")).write_text(source)
    with tempfile.TemporaryDirectory(prefix="mutant_", dir=ROOT) as directory:
        directory = Path(directory)
        (directory / "allocate.py").write_text(source)
        for filename in ("test_allocate.py", "fixtures-input.json", "fixtures-output.json"):
            shutil.copyfile(ROOT / filename, directory / filename)
        probe, probe_data = run([sys.executable, "-c", "import allocate; print(allocate.__file__)"], directory, name + ".import")
        if probe.returncode or Path(probe.stdout.strip()) != directory / "allocate.py":
            raise SystemExit(f"Wrong mutation import: {name}")
        result, record = run([sys.executable, "-m", "unittest", "-v", "test_allocate"], directory, name)
        record["compile_ok"] = True
        record["isolated_import_verified"] = True
        record["assertion_failure_present"] = "FAIL:" in result.stderr
        record["no_test_errors"] = "ERROR:" not in result.stderr
        record["killed"] = (result.returncode != 0 and record["assertion_failure_present"] and record["no_test_errors"])
        report["mutations"][name] = record
report["all_mutations_killed"] = all(record["killed"] for record in report["mutations"].values())
(LOGS / "report.json").write_text(json.dumps(report, indent=2) + "\n")
print(json.dumps(report, indent=2))
if not report["all_mutations_killed"]:
    raise SystemExit("A mutant survived or failed for an invalid reason")