import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import fetch  # noqa: E402
import lens  # noqa: E402
import render  # noqa: E402

BASE = ("Reputation is non-transferable and decays if unused: it is earned only by graded work and drops "
        "after many epochs without a passing answer. Buying an aged account therefore buys something already "
        "decaying, and an attacker's stock of aliases loses value while it sits idle. The {a} ledger stores the "
        "decay curve, the {b} snapshot records each drop, and the {c} score is recomputed from replay.")
OTHER = ("Require each new identity to lock a refundable bond that a randomly drawn jury can burn when it finds "
         "duplicated reasoning, so that running a thousand puppets costs a thousand bonds and every copied answer "
         "risks all of them at once; honest newcomers get the bond back after a probation period of clean work.")
THIRD = ("Rate-limit minting with proof of sequential work: every account must show a verifiable delay proof "
         "computed over many hours, which parallel hardware cannot shortcut, so a farm of ten thousand accounts "
         "needs ten thousand slow proofs while a single honest agent barely notices the one-time wait at join.")


def A(frm, ep, ts, text):
    return {"from": frm, "epoch": ep, "ts": ts, "answer": text}


class TestTokens(unittest.TestCase):
    def test_seed_words_stripped_and_lowercase(self):
        self.assertEqual(lens.tokens("Ember RIVER garden-Path x2y"), ["path", "x", "y"])

    def test_seed_swap_is_containment_one(self):
        a = lens.grams(lens.tokens(BASE.format(a="dawn", b="zenith", c="lotus")))
        b = lens.grams(lens.tokens(BASE.format(a="sonnet", b="juniper", c="quill")))
        self.assertEqual(lens.containment(a, b), 1.0)
        self.assertEqual(lens.jaccard(a, b), 1.0)

    def test_unrelated_low(self):
        a = lens.grams(lens.tokens(OTHER))
        b = lens.grams(lens.tokens(THIRD))
        self.assertLess(lens.containment(a, b), 0.05)

    def test_task_grams_dropped(self):
        g = lens.grams(lens.tokens("one concrete way an agent economy can resist sybil attacks"))
        self.assertEqual(g, set())

    def test_small_set_uses_newer(self):
        a = {(str(i),) for i in range(10)}
        b = {(str(i),) for i in range(5)} | {("x%d" % i,) for i in range(40)}
        # smaller set (a) has < 20 grams -> newer answer's set is the denominator
        self.assertAlmostEqual(lens.containment(a, b, newer="b"), 5 / 45)
        self.assertAlmostEqual(lens.containment(a, b, newer="a"), 5 / 10)


class TestFamilies(unittest.TestCase):
    def setUp(self):
        self.ans = {
            "t1": A("pA", 1, 100, BASE.format(a="dawn", b="zenith", c="lotus")),
            "t2": A("pB", 2, 30000, BASE.format(a="sonnet", b="juniper", c="quill")),
            "t3": A("pC", 2, 30500, OTHER),
            "t4": A("pD", 2, 31000, THIRD),
            "t5": A("pA", 3, 50000, BASE.format(a="opal", b="reef", c="tide")),
        }
        self.rows = lens.prepare(self.ans)

    def test_family_needs_two_accounts(self):
        pairs = lens.s1_pairs(self.rows)
        fams = lens.build_families(self.rows, pairs, use_s2=False)
        self.assertEqual(len(fams), 1)
        f = fams[0]
        self.assertEqual(set(f["authors"]), {"pA", "pB"})
        self.assertEqual(f["earliest"], "t1")
        self.assertTrue(f["registry"])  # epochs 1,2,3
        self.assertEqual(f["id"], lens.family_id("t1"))

    def test_same_author_only_is_not_a_family(self):
        rows = lens.prepare({k: v for k, v in self.ans.items() if k != "t2"})
        fams = lens.build_families(rows, lens.s1_pairs(rows), use_s2=False)
        self.assertEqual(fams, [])

    def test_allowlist_excludes(self):
        fams = lens.build_families(self.rows, lens.s1_pairs(self.rows), use_s2=False, excluded={"pB"})
        self.assertEqual(fams, [])

    def test_threshold(self):
        half = " ".join(BASE.format(a="dawn", b="zenith", c="lotus").split()[:20]) + " " + OTHER
        rows = lens.prepare({"x": A("p1", 1, 1, BASE.format(a="dawn", b="zenith", c="lotus")), "y": A("p2", 1, 2, half)})
        pairs = lens.s1_pairs(rows)
        self.assertEqual(len(pairs), 1)
        self.assertGreaterEqual(pairs[0][2], 0.25)
        rows2 = lens.prepare({"x": A("p1", 1, 1, OTHER), "y": A("p2", 1, 2, THIRD)})
        self.assertEqual(lens.s1_pairs(rows2), [])

    def test_shared_spans_marks_overlap(self):
        ma, mb = lens.shared_spans(self.ans["t1"]["answer"], self.ans["t3"]["answer"])
        self.assertFalse(any(s for _, s in ma))
        ma, mb = lens.shared_spans(self.ans["t1"]["answer"], self.ans["t2"]["answer"])
        self.assertTrue(all(s for w, s in ma[:10]))


class TestS2(unittest.TestCase):
    def test_masked_sentence(self):
        m = lens.mask_sentence_tokens("From Agent0042's view, cost asymmetry is the key defense near the ember")
        self.assertEqual(m[1], "#")
        self.assertEqual(m[-1], "#")

    def test_groups_need_four_answers_two_authors(self):
        s = "In practice Agent%d sees clones fail at the grading gate every single epoch here."
        ans = {"a%d" % i: A("p%d" % (i % 2), 1, i, (s % i) + " " + OTHER[:40]) for i in range(4)}
        g = lens.s2_groups(lens.prepare(ans))
        self.assertIn("in practice # sees clones fail at the grading gate every single epoch here", g)
        ans = {"a%d" % i: A("p0", 1, i, s % i) for i in range(4)}
        self.assertEqual(lens.s2_groups(lens.prepare(ans)), {})


class TestTiming(unittest.TestCase):
    def idx(self, rows):
        return {"t%d" % i: {"from": f, "epoch": e, "ts": ts} for i, (f, e, ts) in enumerate(rows)}

    def test_lockstep_three_epochs(self):
        rows = []
        for e in (1, 2, 3):
            rows += [("pA", e, e * 21600 + 100), ("pB", e, e * 21600 + 103), ("pC", e, e * 21600 + 5000 + e * 700)]
        p = lens.sync_pairs(self.idx(rows))
        self.assertEqual(list(p), [("pA", "pB")])
        self.assertEqual([e for e, _ in p[("pA", "pB")]], [1, 2, 3])

    def test_two_epochs_not_enough(self):
        rows = []
        for e in (1, 2):
            rows += [("pA", e, e * 21600 + 100), ("pB", e, e * 21600 + 101)]
        self.assertEqual(lens.sync_pairs(self.idx(rows)), {})

    def test_busy_moment_skipped(self):
        rows = []
        for e in (1, 2, 3):
            rows += [("p%d" % k, e, e * 21600 + 100 + k) for k in range(6)]
        self.assertEqual(lens.sync_pairs(self.idx(rows)), {})


class TestIntros(unittest.TestCase):
    def test_name_masked_near_duplicate(self):
        body = ("Hello forum, I'm {n}, a language-model agent running on a chat harness, operator-assisted: my human owns "
                "the key and I only ever send it to the network. I read the guides and several recent introductions before "
                "posting, and I mine on an hourly schedule while I learn how grading works here.")
        posts = [{"id": 1, "author": "Alpha", "title": "Alpha intro", "body": body.format(n="Alpha")},
                 {"id": 2, "author": "Beta", "title": "Beta intro", "body": body.format(n="Beta")},
                 {"id": 3, "author": "Gamma", "title": "Gamma", "body": OTHER}]
        pr = lens.intro_pairs(posts)
        self.assertEqual([(a, b) for a, b, _ in pr], [(1, 2)])
        g = lens.intro_groups(posts, pr)
        self.assertEqual(g[0]["posts"], [1, 2])


class TestRenderAndFetch(unittest.TestCase):
    def test_doc_and_html(self):
        ans = {"t1": A("pA", 1, 100, BASE.format(a="dawn", b="zenith", c="lotus")),
               "t2": A("pB", 2, 30000, BASE.format(a="sonnet", b="juniper", c="quill"))}
        idx = {k: {"from": v["from"], "epoch": v["epoch"], "ts": v["ts"]} for k, v in ans.items()}
        doc = render.build_doc(ans, idx, [], {"entries": []}, 1792800000)
        self.assertEqual(doc["summary"]["families"], 1)
        h = render.render_html(doc)
        self.assertIn("clusters worth a look", h)
        self.assertIn("no payout effect", h)
        self.assertIn("<mark>", h)
        for bad in ("sybil farm", "fake account", "operator of", "fakes"):
            self.assertNotIn(bad, h.lower())
        doc2 = render.build_doc(ans, idx, [], {"entries": [{"addresses": ["pB"], "reason": "shared public starter text"}]}, 1792800000)
        self.assertEqual(doc2["summary"]["families"], 0)
        self.assertEqual(doc2["summary"]["cleared"], 1)

    def test_mask_word(self):
        self.assertEqual(render.mask_word("lantern"), "lantern")

    def test_fill_bodies_stops_after_consecutive_errors(self):
        calls = []
        def bad(url, timeout=60):
            calls.append(url); raise OSError("timeout")
        idx = {"t%d" % i: {"epoch": 1, "ts": i} for i in range(10)}
        got, missing, why = fetch.fill_bodies(idx, {}, budget=10, get=bad, sleep=lambda s: None)
        self.assertEqual((got, why, len(calls)), (0, "errors", 3))

    def test_fill_bodies_budget_and_busy(self):
        idx = {"a": {"epoch": 2, "ts": 5}, "b": {"epoch": 1, "ts": 1}, "c": {"epoch": 2, "ts": 9}}
        calls = []

        def get(url, timeout=0):
            calls.append(url)
            t = url.rsplit("/", 1)[1]
            return {"type": "challenge", "from_addr": "p" + t, "epoch": idx[t]["epoch"], "ts": idx[t]["ts"],
                    "body": {"answer": "text " + t, "seed": "00"}}
        bodies = {}
        got, missing, why = fetch.fill_bodies(idx, bodies, budget=2, get=get, sleep=lambda s: None)
        self.assertEqual((got, missing, why), (2, 1, "budget"))
        self.assertEqual(sorted(bodies), ["a", "c"])  # newest epoch first

        def busy(url, timeout=0):
            raise fetch.Busy(60)
        slept = []
        got, missing, why = fetch.fill_bodies(idx, {}, budget=5, get=busy, sleep=slept.append, max_busy=3)
        self.assertEqual((got, why), (0, "node busy"))
        self.assertEqual(slept, [60, 60])

    def test_http_307_busy_is_busy(self):
        import io, urllib.error, email.message
        import fetch
        def opener_raising(code, body):
            class O:
                def open(self, req, timeout=0):
                    h = email.message.Message()
                    raise urllib.error.HTTPError(req.full_url, code, "x", h, io.BytesIO(body))
            return O()
        old = fetch._OPENER
        try:
            fetch._OPENER = opener_raising(307, b'{"error":"origin busy","retry_after_seconds":60}')
            with self.assertRaises(fetch.Busy) as cm:
                fetch.http_get_json("https://example.invalid/x")
            self.assertEqual(cm.exception.retry_after, 60)
            fetch._OPENER = opener_raising(307, b"")
            with self.assertRaises(urllib.error.HTTPError):
                fetch.http_get_json("https://example.invalid/x")
        finally:
            fetch._OPENER = old

    def test_update_index_stops(self):
        pages = [[{"id": "x%d" % i, "type": "challenge", "from_addr": "p", "epoch": 1, "ts": i} for i in range(1000)],
                 [{"id": "y", "type": "challenge", "from_addr": "p", "epoch": 0, "ts": 0}]]
        idx = {}
        n = fetch.update_index(idx, get=lambda u: pages[0] if "offset=0" in u else pages[1], sleep=lambda s: None)
        self.assertEqual(n, 1001)


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


class SegmentTagTest(unittest.TestCase):
    def test_segment_of_and_tagging(self):
        import fetch
        self.assertEqual(fetch.segment_of(1792848488), "poa-mainnet-1-segment1")
        self.assertEqual(fetch.segment_of(1792876000), "poa-mainnet-1-segment2")
        self.assertEqual(fetch.segment_of(None), "poa-mainnet-1-segment1")
        idx = {"a": {"ts": 1792800000, "epoch": 1}, "b": {"ts": 1792880000, "epoch": 1}, "c": {"ts": 5, "seg": "x"}}
        self.assertEqual(fetch.tag_segments(idx), 2)
        self.assertEqual((idx["a"]["seg"], idx["b"]["seg"], idx["c"]["seg"]),
                         ("poa-mainnet-1-segment1", "poa-mainnet-1-segment2", "x"))

    def test_update_index_tags_new_rows(self):
        import fetch
        idx = {}
        rows = [{"type": "challenge", "id": "t1", "from_addr": "poa1x", "epoch": 1, "ts": 1792880000}]
        fetch.update_index(idx, get=lambda url: rows, sleep=lambda s: None)
        self.assertEqual(idx["t1"]["seg"], "poa-mainnet-1-segment2")